List:Commits« Previous MessageNext Message »
From:Alexander Nozdrin Date:December 2 2008 12:06pm
Subject:bzr commit into mysql-6.0-runtime branch (alik:2756) WL#4264
View as plain text  
#At file:///mnt/raid/alik/MySQL/bzr/wl4264/6.0-rt-wl4264/

 2756 Alexander Nozdrin	2008-12-02 [merge]
      Patch for WL#4264: Backup: Stabilize Service Interface.
      
      This patch consists of two major parts:
        - Implementing "execute direct" technique -- a way to execute
          arbitrary SQL statement from within the server and get the
          results without involving transport layer.
        - Implementing Service Interface on top of the execute direct
          technique.
modified:
  libmysql/libmysql.c
  mysql-test/suite/backup/r/backup_db_grants.result
  mysql-test/suite/backup/r/backup_errors.result
  mysql-test/suite/backup/r/backup_lock_myisam.result
  mysql-test/suite/backup/r/backup_views.result
  mysql-test/suite/backup/t/backup_db_grants.test
  mysql-test/suite/backup/t/backup_errors.test
  mysql-test/suite/backup/t/backup_views.test
  sql/backup/backup_aux.h
  sql/backup/backup_info.cc
  sql/backup/backup_info.h
  sql/backup/backup_test.cc
  sql/backup/image_info.h
  sql/backup/kernel.cc
  sql/backup/logger.cc
  sql/debug_sync.cc
  sql/mysql_priv.h
  sql/mysql_priv.h.pp
  sql/protocol.cc
  sql/protocol.h
  sql/share/errmsg.txt
  sql/si_objects.cc
  sql/si_objects.h
  sql/sql_cache.cc
  sql/sql_error.h
  sql/sql_lex.cc
  sql/sql_lex.h
  sql/sql_prepare.cc
  sql/sql_string.h
  sql/sql_yacc.yy

per-file messages:
  libmysql/libmysql.c
    Preserve error message.
  mysql-test/suite/backup/r/backup_db_grants.result
    Update result file.
  mysql-test/suite/backup/r/backup_errors.result
    Update result file.
  mysql-test/suite/backup/r/backup_lock_myisam.result
    Update result file.
  mysql-test/suite/backup/r/backup_views.result
    Update result file.
  mysql-test/suite/backup/t/backup_db_grants.test
    Add additional check to the test case.
  mysql-test/suite/backup/t/backup_errors.test
    Error code has been changed.
  mysql-test/suite/backup/t/backup_views.test
    Error code has been changed.
  sql/backup/backup_aux.h
    Remove unused defines.
  sql/backup/backup_info.cc
    Use new SI.
  sql/backup/backup_info.h
    Use new SI.
  sql/backup/backup_test.cc
    Use new SI.
  sql/backup/image_info.h
    Use new SI.
  sql/backup/kernel.cc
    Use new SI.
  sql/backup/logger.cc
    Fix a subtle bug: if a warning or an error contained '%',
    the log output was invalid.
  sql/debug_sync.cc
    Eliminate compiler warning.
  sql/mysql_priv.h
    Introduce execute direct technique:
      - Server_runnable -- an "executable" inteface.
      - mysql_execute_direct() -- functions to execute
        arbitrary code in the current THD context with
        proper cleanup.
  sql/mysql_priv.h.pp
    Update mysql_priv.h.pp.
  sql/protocol.cc
    In order to support getting results in mysql_execute_direct()
    the following classes were added:
      - Protocol_local
      - Protocol_void
      - Ed_result
      - Ed_result_set
      - Ed_result_set_metadata
      - Ed_row
      - Ed_column
  sql/protocol.h
    In order to support getting results in mysql_execute_direct()
    the following classes were added:
      - Protocol_local
      - Protocol_void
      - Ed_result
      - Ed_result_set
      - Ed_result_set_metadata
      - Ed_row
      - Ed_column
  sql/share/errmsg.txt
    Update error message.
  sql/si_objects.cc
    Refactor SI implementation basing on mysql_execute_direct().
  sql/si_objects.h
    Refactor SI implementation basing on mysql_execute_direct().
  sql/sql_cache.cc
    Support execute direct technique.
  sql/sql_error.h
    Remove default constructor. Warning_info must be initialized properly.
  sql/sql_lex.cc
    Introduce a new flag -- multi_statement. It determines if multiple
    statements are allowed in the input. stmt_prepare_mode flag was used
    previously for that purpose. However, after introduction of
    mysql_execute_direct() (which is not a prepare statement, but multiple
    statements are prohibitted) that needs to be changed.
  sql/sql_prepare.cc
    Implement execute direct technique.
  sql/sql_string.h
    Add an utility method.
  sql/sql_yacc.yy
    Introduce a new flag -- multi_statement. It determines if multiple
    statements are allowed in the input. stmt_prepare_mode flag was used
    previously for that purpose. However, after introduction of
    mysql_execute_direct() (which is not a prepare statement, but multiple
    statements are prohibitted) that needs to be changed.
=== modified file 'libmysql/libmysql.c'
--- a/libmysql/libmysql.c	2008-10-22 11:51:28 +0000
+++ b/libmysql/libmysql.c	2008-11-19 16:32:01 +0000
@@ -4863,7 +4863,10 @@ int STDCALL mysql_stmt_next_result(MYSQL
   rc= mysql_next_result(mysql);
 
   if (rc)
+  {
+    set_stmt_errmsg(stmt, &mysql->net);
     DBUG_RETURN(rc);
+  }
 
   stmt->state= MYSQL_STMT_EXECUTE_DONE;
   stmt->bind_result_done= FALSE;

=== modified file 'mysql-test/suite/backup/r/backup_db_grants.result'
--- a/mysql-test/suite/backup/r/backup_db_grants.result	2008-10-07 17:15:44 +0000
+++ b/mysql-test/suite/backup/r/backup_db_grants.result	2008-11-19 16:32:01 +0000
@@ -63,10 +63,31 @@ GRANT USAGE ON *.* TO 'bup_user2'@'%'
 SHOW GRANTS FOR 'bup_user3'@'%';
 Grants for bup_user3@%
 GRANT USAGE ON *.* TO 'bup_user3'@'%'
+SHOW GRANTS FOR 'no_user'@'%';
+ERROR 42000: There is no such grant defined for user 'no_user' on host '%'
 Run Restore
 RESTORE FROM 'bup_db_grants.bak';
 backup_id
 #
+Warnings:
+#	1737	The grant 'ALTER ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'ALTER ROUTINE ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'CREATE ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'CREATE ROUTINE ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'CREATE TEMPORARY TABLES ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'CREATE VIEW ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'DELETE ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'DROP ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'EVENT ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'EXECUTE ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'INDEX ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'INSERT ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'LOCK TABLES ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'REFERENCES ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'SELECT ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'SHOW VIEW ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'TRIGGER ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'UPDATE ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
 SHOW TABLES FROM bup_db_grants;
 Tables_in_bup_db_grants
 s1
@@ -99,6 +120,28 @@ Run Restore
 RESTORE FROM 'bup_db_grants.bak';
 backup_id
 #
+Warnings:
+#	1737	The grant 'INSERT ON bup_db_grants.*' for the user 'bup_user2'@'%' was skipped because the user does not exist.
+#	1737	The grant 'INSERT(b) ON bup_db_grants.s1' for the user 'bup_user2'@'%' was skipped because the user does not exist.
+#	1737	The grant 'SELECT(b) ON bup_db_grants.s1' for the user 'bup_user2'@'%' was skipped because the user does not exist.
+#	1737	The grant 'ALTER ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'ALTER ROUTINE ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'CREATE ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'CREATE ROUTINE ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'CREATE TEMPORARY TABLES ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'CREATE VIEW ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'DELETE ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'DROP ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'EVENT ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'EXECUTE ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'INDEX ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'INSERT ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'LOCK TABLES ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'REFERENCES ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'SELECT ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'SHOW VIEW ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'TRIGGER ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
+#	1737	The grant 'UPDATE ON bup_db_grants.*' for the user 'no_user'@'%' was skipped because the user does not exist.
 SHOW TABLES FROM bup_db_grants;
 Tables_in_bup_db_grants
 s1

=== modified file 'mysql-test/suite/backup/r/backup_errors.result'
--- a/mysql-test/suite/backup/r/backup_errors.result	2008-10-07 17:15:44 +0000
+++ b/mysql-test/suite/backup/r/backup_errors.result	2008-11-19 16:32:01 +0000
@@ -110,7 +110,7 @@ delete database files so that check_db_d
 si_objects.cc @ DatabaseObj::do_serialize
 SET DEBUG_SYNC='now SIGNAL db_will_fail';
 Database has been deleted, backup will fail
-ERROR 42000: Unknown database 'db1'
+ERROR HY000: Failed to obtain meta-data for database `db1`
 
 Test that backup fails with error if a table used by 
 a trigger cannot be opened

=== modified file 'mysql-test/suite/backup/r/backup_lock_myisam.result'
--- a/mysql-test/suite/backup/r/backup_lock_myisam.result	2008-10-07 17:15:44 +0000
+++ b/mysql-test/suite/backup/r/backup_lock_myisam.result	2008-11-28 21:35:11 +0000
@@ -61,6 +61,10 @@ SET DEBUG_SYNC= 'now WAIT_FOR wait_for_r
 breakpoints: Show process list.
 SELECT id, command, state, info FROM INFORMATION_SCHEMA.PROCESSLIST
 WHERE info LIKE "RESTORE%";
+id	#
+command	Query
+state	debug sync point: restore_in_progress
+info	RESTORE FROM 'db1.bak'
 From con2:
 Now do the insert while restore is running.
 INSERT INTO db2.t2 VALUES (0);
@@ -99,6 +103,10 @@ SET DEBUG_SYNC= 'now WAIT_FOR wait_for_r
 breakpoints: Show process list.
 SELECT id, command, state, info FROM INFORMATION_SCHEMA.PROCESSLIST
 WHERE info LIKE "RESTORE%";
+id	#
+command	Query
+state	debug sync point: restore_in_progress
+info	RESTORE FROM 'db1.bak'
 From con2:
 Now do the select while restore is running.
 SELECT * FROM db1.t1 limit 10;

=== modified file 'mysql-test/suite/backup/r/backup_views.result'
--- a/mysql-test/suite/backup/r/backup_views.result	2008-10-07 17:15:44 +0000
+++ b/mysql-test/suite/backup/r/backup_views.result	2008-11-26 13:46:42 +0000
@@ -584,7 +584,7 @@ Testing view selecting from altered view
 SELECT * FROM alter2;
 ERROR HY000: View 'bup_db1.alter2' references invalid table(s) or column(s) or function(s) or definer/invoker of view lack rights to use them
 BACKUP DATABASE bup_db1 TO 'bup_alterview.bak';
-ERROR HY000: View 'bup_db1.alter2' references invalid table(s) or column(s) or function(s) or definer/invoker of view lack rights to use them
+ERROR HY000: Failed to obtain meta-data for view `bup_db1`.`alter2`
 
 *** EXIT Backup of database with altered view
 

=== modified file 'mysql-test/suite/backup/t/backup_db_grants.test'
--- a/mysql-test/suite/backup/t/backup_db_grants.test	2008-10-07 17:15:44 +0000
+++ b/mysql-test/suite/backup/t/backup_db_grants.test	2008-11-19 16:32:01 +0000
@@ -60,6 +60,8 @@ FLUSH PRIVILEGES;
 SHOW GRANTS FOR 'bup_user1'@'%';
 SHOW GRANTS FOR bup_user2;
 SHOW GRANTS FOR 'bup_user3'@'%';
+--error ER_NONEXISTING_GRANT
+SHOW GRANTS FOR 'no_user'@'%';
 
 --echo Run Restore
 --replace_column 1 #

=== modified file 'mysql-test/suite/backup/t/backup_errors.test'
--- a/mysql-test/suite/backup/t/backup_errors.test	2008-10-07 17:15:44 +0000
+++ b/mysql-test/suite/backup/t/backup_errors.test	2008-11-19 16:32:01 +0000
@@ -251,7 +251,7 @@ SET DEBUG_SYNC='now SIGNAL db_will_fail'
 
 connection con2;
 --echo Database has been deleted, backup will fail
---error ER_BAD_DB_ERROR
+--error ER_BACKUP_GET_META_DB
 reap;
 
 --echo

=== modified file 'mysql-test/suite/backup/t/backup_views.test'
--- a/mysql-test/suite/backup/t/backup_views.test	2008-10-07 17:15:44 +0000
+++ b/mysql-test/suite/backup/t/backup_views.test	2008-11-26 13:46:42 +0000
@@ -353,7 +353,7 @@ ALTER VIEW alter1 AS SELECT 6;
 SELECT * FROM alter2;
 
 #fails
---error ER_VIEW_INVALID
+--error ER_BACKUP_GET_META_VIEW
 BACKUP DATABASE bup_db1 TO 'bup_alterview.bak';
 
 --echo 

=== modified file 'sql/backup/backup_aux.h'
--- a/sql/backup/backup_aux.h	2008-10-14 12:08:56 +0000
+++ b/sql/backup/backup_aux.h	2008-11-19 16:32:01 +0000
@@ -42,12 +42,6 @@ storage_engine_ref get_se_by_name(const 
 namespace backup {
 
 /**
-  Constants for appending uniqueness to privileges in backup catalog.
-*/
-#define UNIQUE_PRIV_KEY_LEN 9
-#define UNIQUE_PRIV_KEY_FORMAT "%08lu"
-
-/**
   Local version of LEX_STRING structure.
 
   Defines various constructors for convenience.

=== modified file 'sql/backup/backup_info.cc'
--- a/sql/backup/backup_info.cc	2008-10-27 13:06:21 +0000
+++ b/sql/backup/backup_info.cc	2008-11-25 16:37:46 +0000
@@ -532,7 +532,7 @@ backup::Image_info::Db* Backup_info::add
 
   @returns 0 on success, error code otherwise.
  */
-int Backup_info::add_dbs(List< ::LEX_STRING > &dbs)
+int Backup_info::add_dbs(THD *thd, List< ::LEX_STRING > &dbs)
 {
   using namespace obs;
 
@@ -550,9 +550,9 @@ int Backup_info::add_dbs(List< ::LEX_STR
       goto error;
     }
     
-    obs::Obj *obj= get_database(&db_name); // reports errors
+    obs::Obj *obj= get_database_stub(&db_name); // reports errors
 
-    if (obj && !check_db_existence(&db_name))
+    if (obj && !check_db_existence(thd, &db_name))
     {    
       if (!unknown_dbs.is_empty()) // we just compose unknown_dbs list
       {
@@ -727,7 +727,7 @@ int Backup_info::add_db_items(Db &db)
 
     // If this table uses a tablespace, add this tablespace to the catalogue.
 
-    obj= get_tablespace_for_table(m_ctx.m_thd, &db.name(), &tbl->name());
+    obj= find_tablespace_for_table(m_ctx.m_thd, &db.name(), &tbl->name());
 
     if (obj)
     {
@@ -1031,8 +1031,7 @@ Backup_info::add_db_object(Db &db, const
   ulong pos= db.obj_count();
 
   DBUG_ASSERT(obj);
-  String *name= (String *)obj->get_name();
-  DBUG_ASSERT(name);
+  DBUG_ASSERT(obj->get_name());
 
   switch (type) {
 
@@ -1052,22 +1051,6 @@ Backup_info::add_db_object(Db &db, const
 
   }
 
-  /*
-    Generate a unique name for the privilege (grant) objects.
-    Note: this name does not alter the mechanics of the
-          grant objects in si_objects.cc
-  */
-  if (type == BSTREAM_IT_PRIVILEGE)
-  {
-    String new_name;
-    char buff[10];
-    sprintf(buff, UNIQUE_PRIV_KEY_FORMAT, pos);
-    new_name.append(*name);
-    new_name.append(" ");
-    new_name.append(buff);
-    name->copy(new_name);
-  }
-
   /* 
     Add new object to the dependency list. If it is a view, add its
     dependencies first.
@@ -1078,11 +1061,11 @@ Backup_info::add_db_object(Db &db, const
 
   // Get a dep. list node for the object.  
 
-  int res= get_dep_node(db.name(), *name, type, n);
+  int res= get_dep_node(db.name(), *obj->get_name(), type, n);
   
   if (res == get_dep_node_res::ERROR)
   {
-    m_ctx.fatal_error(error, db.name().ptr(), name->ptr());
+    m_ctx.fatal_error(error, db.name().ptr(), obj->get_name()->ptr());
     return NULL;
   }
 
@@ -1097,7 +1080,7 @@ Backup_info::add_db_object(Db &db, const
     if (type == BSTREAM_IT_VIEW)
       if (add_view_deps(*obj))
       {
-        m_ctx.fatal_error(error, db.name().ptr(), name->ptr());
+        m_ctx.fatal_error(error, db.name().ptr(), obj->get_name()->ptr());
         return NULL;
       } 
 
@@ -1114,11 +1097,11 @@ Backup_info::add_db_object(Db &db, const
     objects.
    */
 
-  Dbobj *o= Image_info::add_db_object(db, type, *name, pos);
+  Dbobj *o= Image_info::add_db_object(db, type, *obj->get_name(), pos);
  
   if (!o)
   {
-    m_ctx.fatal_error(error, db.name().ptr(), name->ptr());
+    m_ctx.fatal_error(error, db.name().ptr(), obj->get_name()->ptr());
     return NULL;
   }
 
@@ -1134,7 +1117,7 @@ Backup_info::add_db_object(Db &db, const
   n->obj= o;  
 
   DBUG_PRINT("backup",("Added object %s of type %d from database %s (pos=%lu)",
-                       name->ptr(), type, db.name().ptr(), pos));
+                       obj->get_name()->ptr(), type, db.name().ptr(), pos));
   return o;
 }
 

=== modified file 'sql/backup/backup_info.h'
--- a/sql/backup/backup_info.h	2008-10-27 13:06:21 +0000
+++ b/sql/backup/backup_info.h	2008-11-19 16:32:01 +0000
@@ -38,7 +38,7 @@ class Backup_info: public backup::Image_
 
   bool is_valid();
 
-  int add_dbs(List< ::LEX_STRING >&);
+  int add_dbs(THD *thd, List< ::LEX_STRING >&);
   int add_all_dbs();
 
   int close();

=== modified file 'sql/backup/backup_test.cc'
--- a/sql/backup/backup_test.cc	2008-10-17 11:28:25 +0000
+++ b/sql/backup/backup_test.cc	2008-11-25 16:37:46 +0000
@@ -32,7 +32,7 @@ int execute_backup_test_command(THD *thd
 
   {
     String tmp_db_name("qqq", 3, system_charset_info);
-    DBUG_ASSERT(obs::check_db_existence(&tmp_db_name));
+    DBUG_ASSERT(obs::check_db_existence(thd, &tmp_db_name));
   }
 
   /*
@@ -57,12 +57,12 @@ int execute_backup_test_command(THD *thd
     {
       String dir;
       dir.copy(dbname->str, dbname->length, system_charset_info);
-      db= get_database(&dir);
+      db= get_database_stub(&dir);
 
       if (is_internal_db_name(db->get_db_name()))
           continue;
 
-      DBUG_ASSERT(!obs::check_db_existence(db->get_db_name()));
+      DBUG_ASSERT(!obs::check_db_existence(thd, db->get_db_name()));
 
       //
       // List tables..
@@ -247,10 +247,7 @@ int execute_backup_test_command(THD *thd
           protocol->prepare_for_resend();
           protocol->store(const_cast<String*>(db->get_name()));
           protocol->store(const_cast<String*>(grant->get_name()));
-          String user;
-          String host;
-          String *user_host= (String *)grant->get_name();
-          check_user_existence(thd, user_host);
+          check_user_existence(thd, grant);
           protocol->store(C_STRING_WITH_LEN("GRANT"),
                           system_charset_info);
           grant->serialize(thd, &serial);

=== modified file 'sql/backup/image_info.h'
--- a/sql/backup/image_info.h	2008-10-15 15:38:28 +0000
+++ b/sql/backup/image_info.h	2008-11-25 16:37:46 +0000
@@ -1021,7 +1021,7 @@ inline
 obs::Obj* Image_info::Ts::materialize(uint ver, const ::String &sdata)
 {
   delete m_obj_ptr;
-  return m_obj_ptr= obs::materialize_tablespace(&m_name, ver, &sdata); 
+  return m_obj_ptr= obs::get_tablespace(&m_name, ver, &sdata); 
 }
 
 /// Implementation of @c Image_info::Obj virtual method.
@@ -1029,7 +1029,7 @@ inline
 obs::Obj* Image_info::Db::materialize(uint ver, const ::String &sdata)
 {
   delete m_obj_ptr;
-  return m_obj_ptr= obs::materialize_database(&name(), ver, &sdata); 
+  return m_obj_ptr= obs::get_database(&name(), ver, &sdata); 
 }
 
 /// Implementation of @c Image_info::Obj virtual method.
@@ -1037,7 +1037,7 @@ inline
 obs::Obj* Image_info::Table::materialize(uint ver, const ::String &sdata)
 {
   delete m_obj_ptr;
-  return m_obj_ptr= obs::materialize_table(&db().name(), &name(), ver, &sdata);
+  return m_obj_ptr= obs::get_table(&db().name(), &name(), ver, &sdata);
 }
 
 inline
@@ -1050,31 +1050,23 @@ obs::Obj* Image_info::Dbobj::materialize
   
   switch (base.type) {
   case BSTREAM_IT_VIEW:   
-    m_obj_ptr= obs::materialize_view(db_name, name, ver, &sdata);
+    m_obj_ptr= obs::get_view(db_name, name, ver, &sdata);
     break;
   case BSTREAM_IT_SPROC:  
-    m_obj_ptr= obs::materialize_stored_procedure(db_name, name, ver, &sdata);
+    m_obj_ptr= obs::get_stored_procedure(db_name, name, ver, &sdata);
     break;
   case BSTREAM_IT_SFUNC:
-    m_obj_ptr= obs::materialize_stored_function(db_name, name, ver, &sdata); 
+    m_obj_ptr= obs::get_stored_function(db_name, name, ver, &sdata); 
     break;
   case BSTREAM_IT_EVENT:
-    m_obj_ptr= obs::materialize_event(db_name, name, ver, &sdata);
+    m_obj_ptr= obs::get_event(db_name, name, ver, &sdata);
     break;
   case BSTREAM_IT_TRIGGER:   
-    m_obj_ptr= obs::materialize_trigger(db_name, name, ver, &sdata);
+    m_obj_ptr= obs::get_trigger(db_name, name, ver, &sdata);
     break;
   case BSTREAM_IT_PRIVILEGE:
-  {
-    /*
-      Here we undo the uniqueness suffix for grants.
-    */
-    String new_name;
-    new_name.copy(*name);
-    new_name.length(new_name.length() - UNIQUE_PRIV_KEY_LEN);
-    m_obj_ptr= obs::materialize_db_grant(db_name, &new_name, ver, &sdata);
+    m_obj_ptr= obs::get_db_grant(db_name, name, ver, &sdata);
     break;
-  }
   default: m_obj_ptr= NULL;
   }
 

=== modified file 'sql/backup/kernel.cc'
--- a/sql/backup/kernel.cc	2008-11-18 19:41:51 +0000
+++ b/sql/backup/kernel.cc	2008-11-25 16:37:46 +0000
@@ -185,7 +185,8 @@ execute_backup_command(THD *thd, LEX *le
     else
     {
       context.write_message(log_level::INFO, "Backing up selected databases");
-      res= info->add_dbs(lex->db_list); // backup databases specified by user
+      /* Backup databases specified by user. */
+      res= info->add_dbs(thd, lex->db_list);
     }
 
     info->close(); // close catalogue after filling it with objects to backup
@@ -1117,7 +1118,7 @@ int Backup_restore_ctx::restore_triggers
       
       case BSTREAM_IT_TRIGGER:
         DBUG_ASSERT(obj->m_obj_ptr);
-        if (obj->m_obj_ptr->execute(m_thd))
+        if (obj->m_obj_ptr->create(m_thd))
         {
           delete it;
           delete dbit;
@@ -1140,7 +1141,7 @@ int Backup_restore_ctx::restore_triggers
   Image_info::Obj *ev;
 
   while ((ev= it++)) 
-    if (ev->m_obj_ptr->execute(m_thd))
+    if (ev->m_obj_ptr->create(m_thd))
     {
       fatal_error(ER_BACKUP_CANT_RESTORE_EVENT,ev->describe(buf));
       DBUG_RETURN(m_error);
@@ -1905,24 +1906,30 @@ int bcat_create_item(st_bstream_image_he
 
   if (item->type == BSTREAM_IT_TABLESPACE)
   {
-    // if the tablespace exists, there is nothing more to do
-    if (obs::tablespace_exists(thd, sobj))
+    Obj *ts= obs::find_tablespace(thd, sobj->get_name());
+
+    if (ts)
     {
-      DBUG_PRINT("restore",(" skipping tablespace which exists"));
-      return BSTREAM_OK;
-    }
+      /*
+        A tablespace with the same name exists. We have to check if other
+        attributes are the same as they were.
+      */
 
-    /* 
-      If there is a different tablespace with the same name then we can't 
-      re-create the original tablespace used by tables being restored. We report 
-      this and cancel restore process.
-    */ 
+      if (obs::compare_tablespace_attributes(ts, sobj))
+      {
+        /* The tablespace is the same. There is nothing more to do. */
+        DBUG_PRINT("restore",(" skipping tablespace which exists"));
+        return BSTREAM_OK;
+      }
 
-    Obj *ts= obs::is_tablespace(thd, sobj); 
+      /*
+        A tablespace with the same name exists, but it has been changed
+        since backup.  We can't re-create the original tablespace used by
+        tables being restored. We report this and cancel restore process.
+      */
 
-    if (ts)
-    {
-      DBUG_PRINT("restore",(" tablespace has changed on the server - aborting"));
+      DBUG_PRINT("restore",
+                 (" tablespace has changed on the server - aborting"));
       info->m_ctx.fatal_error(ER_BACKUP_TS_CHANGE, desc);
       return BSTREAM_ERROR;
     }
@@ -1943,12 +1950,13 @@ int bcat_create_item(st_bstream_image_he
             error handling work in WL#4384 with possible implementation
             via a related bug report.
     */
-    if (!obs::check_user_existence(thd, sobj->get_name()))
+    if (!obs::check_user_existence(thd, sobj))
     {
-      info->m_ctx.report_error(log_level::WARNING, 
+      info->m_ctx.report_error(log_level::WARNING,
                                ER_BACKUP_GRANT_SKIPPED,
-                               create_stmt);
-      return BSTREAM_OK; 
+                               obs::grant_get_grant_info(sobj)->ptr(),
+                               obs::grant_get_user_name(sobj)->ptr());
+      return BSTREAM_OK;
     }
     /*
       We need to check the grant against the database list to ensure the
@@ -1972,7 +1980,7 @@ int bcat_create_item(st_bstream_image_he
     }
   }
 
-  if (sobj->execute(thd))
+  if (sobj->create(thd))
   {
     info->m_ctx.fatal_error(create_err, desc);
     return BSTREAM_ERROR;

=== modified file 'sql/backup/logger.cc'
--- a/sql/backup/logger.cc	2008-10-27 13:06:21 +0000
+++ b/sql/backup/logger.cc	2008-11-19 16:32:01 +0000
@@ -52,10 +52,10 @@ int Logger::write_message(log_level::val
        error.msg= sql_strdup(msg);
      }
 
-     sql_print_error(out);
+     sql_print_error("%s", out);
      if (m_push_errors)
        push_warning_printf(current_thd, MYSQL_ERROR::WARN_LEVEL_WARN,
-			   error_code, msg);
+			   error_code, "%s", msg);
      DBUG_PRINT("backup_log",("[ERROR] %s", out));
      
      if (m_state == READY || m_state == RUNNING)
@@ -69,15 +69,15 @@ int Logger::write_message(log_level::val
      return 0;
 
    case log_level::WARNING:
-     sql_print_warning(out);
+     sql_print_warning("%s", out);
      if (m_push_errors)
        push_warning_printf(current_thd, MYSQL_ERROR::WARN_LEVEL_WARN,
-                           error_code, msg);
+                           error_code, "%s", msg);
      DBUG_PRINT("backup_log",("[Warning] %s", out));
      return 0;
 
    case log_level::INFO:
-     sql_print_information(out);
+     sql_print_information("%s", out);
      DBUG_PRINT("backup_log",("[Info] %s", out));
      return 0;
 

=== modified file 'sql/debug_sync.cc'
--- a/sql/debug_sync.cc	2008-09-08 09:25:46 +0000
+++ b/sql/debug_sync.cc	2008-11-20 11:01:12 +0000
@@ -832,7 +832,7 @@ static st_debug_sync_action *debug_sync_
     /* Reuse an already active sync point action. */
     DBUG_ASSERT((uint)(action - ds_control->ds_action) < ds_control->ds_active);
     DBUG_PRINT("debug_sync", ("reuse action idx: %ld",
-                              action - ds_control->ds_action));
+                              (long) (action - ds_control->ds_action)));
   }
   else
   {

=== modified file 'sql/mysql_priv.h'
--- a/sql/mysql_priv.h	2008-11-06 18:39:27 +0000
+++ b/sql/mysql_priv.h	2008-11-28 21:35:11 +0000
@@ -922,7 +922,7 @@ struct Query_cache_query_flags
 {
   unsigned int client_long_flag:1;
   unsigned int client_protocol_41:1;
-  unsigned int result_in_binary_protocol:1;
+  unsigned int protocol_type:2;
   unsigned int more_results_exists:1;
   unsigned int pkt_nr;
   uint character_set_client_num;
@@ -1136,6 +1136,26 @@ void init_update_queries(void);
 void free_max_user_conn(void);
 pthread_handler_t handle_bootstrap(void *arg);
 int mysql_execute_command(THD *thd);
+
+class Ed_result;
+/**
+  Execute a fragment of server code in an isolated context, so that
+  it doesn't leave any effect on THD. THD must have no open tables.
+  The code must not leave any open tables around.
+  The result of execution (if any) is stored in Ed_result.
+*/
+
+class Server_runnable
+{
+public:
+  virtual bool execute_server_code(THD *thd)= 0;
+  virtual ~Server_runnable();
+};
+
+bool mysql_execute_direct(THD *thd, LEX_STRING query, Ed_result *result);
+bool mysql_execute_direct(THD *thd, Server_runnable *ed_runnable,
+                          Ed_result *result);
+
 bool do_command(THD *thd);
 bool dispatch_command(enum enum_server_command command, THD *thd,
 		      char* packet, uint packet_length);

=== modified file 'sql/mysql_priv.h.pp'
--- a/sql/mysql_priv.h.pp	2008-07-11 13:08:06 +0000
+++ b/sql/mysql_priv.h.pp	2008-11-19 16:32:01 +0000
@@ -7672,7 +7672,6 @@ public:
 };
 void send_warning(THD *thd, uint sql_errno, const char *err=0);
 void net_send_error(THD *thd, uint sql_errno=0, const char *err=0);
-void net_end_statement(THD *thd);
 In_C_you_should_use_my_bool_instead() send_old_password_request(THD *thd);
 uchar *net_store_data(uchar *to,const uchar *from, size_t length);
 uchar *net_store_data(uchar *to,int32 from);

=== modified file 'sql/protocol.cc'
--- a/sql/protocol.cc	2008-11-18 20:25:51 +0000
+++ b/sql/protocol.cc	2008-11-20 11:01:12 +0000
@@ -450,26 +450,26 @@ void Protocol::end_statement()
   switch (thd->main_da.status()) {
   case Diagnostics_area::DA_ERROR:
     /* The query failed, send error to log and abort bootstrap. */
-    this->send_error(thd->main_da.sql_errno(),
-                     thd->main_da.message());
+    send_error(thd->main_da.sql_errno(),
+               thd->main_da.message());
     break;
   case Diagnostics_area::DA_EOF:
-    this->send_eof(thd->main_da.server_status(),
-                   thd->main_da.statement_warn_count());
+    send_eof(thd->main_da.server_status(),
+             thd->main_da.statement_warn_count());
     break;
   case Diagnostics_area::DA_OK:
-    this->send_ok(thd->main_da.server_status(),
-                  thd->main_da.statement_warn_count(),
-                  thd->main_da.affected_rows(),
-                  thd->main_da.last_insert_id(),
-                  thd->main_da.message());
+    send_ok(thd->main_da.server_status(),
+            thd->main_da.statement_warn_count(),
+            thd->main_da.affected_rows(),
+            thd->main_da.last_insert_id(),
+            thd->main_da.message());
     break;
   case Diagnostics_area::DA_DISABLED:
     break;
   case Diagnostics_area::DA_EMPTY:
   default:
     DBUG_ASSERT(0);
-    this->send_ok(thd->server_status, 0, 0, 0, NULL);
+    send_ok(thd->server_status, 0, 0, 0, NULL);
     break;
   }
   thd->main_da.is_sent= TRUE;
@@ -858,7 +858,6 @@ bool Protocol::store(I_List<i_string>* s
   return store((char*) tmp.ptr(), len,  tmp.charset());
 }
 
-
 /****************************************************************************
   Functions to handle the simple (default) protocol where everything is
   This protocol is the one that is used by default between the MySQL server
@@ -1475,3 +1474,477 @@ bool Protocol_binary::send_out_parameter
 
   return FALSE;
 }
+
+///////////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////
+
+Ed_result::Ed_result() :
+  m_current_result_set(NULL),
+  m_status(Diagnostics_area::DA_EMPTY),
+  m_server_status(0),
+  m_affected_rows(0),
+  m_last_insert_id(0),
+  m_sql_errno(0),
+  m_warning_info(0),
+  m_warning_info_saved(NULL)
+{
+  init_sql_alloc(&m_mem_root, ALLOC_ROOT_MIN_BLOCK_SIZE, 0);
+
+  m_message[0]= 0;
+}
+
+///////////////////////////////////////////////////////////////////////////
+
+Ed_result::~Ed_result()
+{
+  free_root(&m_mem_root, MYF(0));
+}
+
+///////////////////////////////////////////////////////////////////////////
+
+bool Ed_result::add_result_set(List<Item> *col_metadata)
+{
+  Ed_result_set *rs= Ed_result_set::create(&m_mem_root, col_metadata);
+
+  if (!rs)
+    return TRUE;
+
+  m_current_result_set= rs;
+
+  return push_back(rs, &m_mem_root) ? TRUE : FALSE;
+}
+
+///////////////////////////////////////////////////////////////////////////
+
+void Ed_result::send_ok(THD *thd,
+                        uint server_status, uint statement_warn_count,
+                        ha_rows affected_rows, ulonglong last_insert_id,
+                        const char *message)
+{
+  DBUG_ENTER("Ed_result::send_ok()");
+  DBUG_ASSERT(m_status == Diagnostics_area::DA_EMPTY);
+
+  m_status= thd->main_da.status();
+  DBUG_ASSERT(m_status == Diagnostics_area::DA_OK);
+
+  DBUG_ASSERT(m_warning_info.statement_warn_count() == statement_warn_count);
+
+  m_server_status= server_status;
+  m_affected_rows= affected_rows;
+  m_last_insert_id= last_insert_id;
+
+  strmake(m_message, message, sizeof (m_message) - 1);
+
+  DBUG_VOID_RETURN;
+}
+
+///////////////////////////////////////////////////////////////////////////
+
+void Ed_result::send_eof(THD *thd, uint server_status,
+                         uint statement_warn_count)
+{
+  DBUG_ENTER("Ed_result::send_eof");
+  DBUG_ASSERT(m_status == Diagnostics_area::DA_EMPTY);
+
+  m_status= thd->main_da.status();
+  DBUG_ASSERT(m_status == Diagnostics_area::DA_EOF);
+
+  DBUG_ASSERT(m_warning_info.statement_warn_count() == statement_warn_count);
+
+  m_server_status= server_status;
+
+  DBUG_VOID_RETURN;
+}
+
+///////////////////////////////////////////////////////////////////////////
+
+void Ed_result::send_error(THD *thd, uint sql_errno, const char *err_msg)
+{
+  DBUG_ENTER("Ed_result::send_error()");
+  DBUG_ASSERT(m_status == Diagnostics_area::DA_EMPTY);
+
+  m_status= thd->main_da.status();
+  DBUG_ASSERT(m_status == Diagnostics_area::DA_ERROR);
+
+  m_sql_errno= sql_errno;
+  strmake(m_message, err_msg, sizeof (m_message) - 1);
+
+  DBUG_VOID_RETURN;
+}
+
+///////////////////////////////////////////////////////////////////////////
+
+void Ed_result::begin_statement(THD *thd)
+{
+  DBUG_ASSERT(!m_warning_info_saved);
+
+  m_warning_info_saved= thd->warning_info;
+  thd->warning_info= &m_warning_info;
+}
+
+///////////////////////////////////////////////////////////////////////////
+
+void Ed_result::end_statement(THD *thd)
+{
+  DBUG_ASSERT(m_warning_info_saved);
+  thd->warning_info= m_warning_info_saved;
+}
+
+///////////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////
+
+Ed_result_set *
+Ed_result_set::create(MEM_ROOT *mem_root, List<Item> *col_metadata)
+{
+  Ed_result_set *rs= new (mem_root) Ed_result_set(mem_root);
+
+  if (!rs || rs->init(col_metadata))
+    return NULL;
+
+  return rs;
+}
+
+///////////////////////////////////////////////////////////////////////////
+
+bool Ed_result_set::init(List<Item> *col_metadata)
+{
+  m_metadata= Ed_result_set_metadata::create(m_mem_root, col_metadata);
+
+  return m_metadata == NULL;
+}
+
+///////////////////////////////////////////////////////////////////////////
+
+Ed_row *Ed_result_set::add_row()
+{
+  Ed_row *row= Ed_row::create(m_mem_root, m_metadata);
+
+  if (!row)
+    return NULL;
+
+  m_current_row= row;
+
+  return m_data.push_back(row, m_mem_root) ? NULL : row;
+}
+
+///////////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////
+
+Ed_result_set_metadata *
+Ed_result_set_metadata::create(MEM_ROOT *mem_root,
+                               List<Item> *col_metadata)
+{
+  Ed_result_set_metadata *md= new (mem_root) Ed_result_set_metadata();
+
+  if (!md || md->init(mem_root, col_metadata))
+    return NULL;
+
+  return md;
+}
+
+///////////////////////////////////////////////////////////////////////////
+
+bool Ed_result_set_metadata::init(MEM_ROOT *mem_root, List<Item> *col_metadata)
+{
+  if (!col_metadata)
+    return FALSE;
+
+  m_metadata= new (mem_root) Send_field[col_metadata->elements];
+
+  if (!m_metadata)
+    return TRUE;
+
+  m_num_columns= col_metadata->elements;
+
+  List_iterator_fast<Item> it(*col_metadata);
+
+  int i= 0;
+  for (Item *column= it++; column; column= it++)
+    column->make_field(&m_metadata[i]);
+
+  return FALSE;
+}
+
+///////////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////
+
+Ed_row *Ed_row::create(MEM_ROOT *mem_root,
+                       const Ed_result_set_metadata *metadata)
+{
+  DBUG_ASSERT(metadata);
+
+  Ed_row *row= new (mem_root) Ed_row(mem_root, metadata);
+
+  if (!row || row->init())
+    return NULL;
+
+  return row;
+}
+
+///////////////////////////////////////////////////////////////////////////
+
+bool Ed_row::init()
+{
+  m_columns= new (m_mem_root) Ed_column[m_metadata->get_num_columns()];
+
+  return m_columns == NULL;
+}
+
+///////////////////////////////////////////////////////////////////////////
+
+bool Ed_row::add_null()
+{
+  if (m_current_column_index >= m_metadata->get_num_columns())
+    return TRUE;
+
+  ++m_current_column_index;
+
+  return FALSE;
+}
+
+///////////////////////////////////////////////////////////////////////////
+
+bool Ed_row::add_column(const void *data_ptr, int data_length)
+{
+  if (m_current_column_index >= m_metadata->get_num_columns())
+    return TRUE;
+
+  m_columns[m_current_column_index++].set_data(m_mem_root,
+                                               data_ptr, data_length);
+
+  return FALSE;
+}
+
+///////////////////////////////////////////////////////////////////////////
+//
+// Protocol_local: a protocol for retrieving result sets from the server
+// locally.
+//
+///////////////////////////////////////////////////////////////////////////
+
+void Protocol_local::prepare_for_resend()
+{
+  DBUG_ASSERT(m_result);
+
+  Ed_result_set *rs= m_result->get_cur_result_set();
+
+  DBUG_ASSERT(rs);
+
+  rs->add_row();
+}
+
+bool Protocol_local::write()
+{
+  return FALSE;
+}
+
+bool Protocol_local::store_null()
+{
+  DBUG_ASSERT(m_result);
+
+  Ed_result_set *rs= m_result->get_cur_result_set();
+  Ed_row *row= rs->get_cur_row();
+  DBUG_ASSERT(row);
+
+  return row->add_null();
+}
+
+bool Protocol_local::store_string(const char *str,
+                                  int length,
+                                  CHARSET_INFO *src_cs,
+                                  CHARSET_INFO *dst_cs)
+{
+  DBUG_ASSERT(m_result);
+
+  Ed_result_set *rs= m_result->get_cur_result_set();
+  Ed_row *row= rs->get_cur_row();
+  DBUG_ASSERT(row);
+
+  /* 'dst_cs' is set 0 when client issued SET character_set_results = NULL */
+
+  if (!dst_cs ||
+      my_charset_same(src_cs, dst_cs) ||
+      src_cs == &my_charset_bin ||
+      dst_cs == &my_charset_bin)
+  {
+    return row->add_column(str, length);
+  }
+
+  /* Store with conversion */
+  uint dummy_errors;
+
+  if (convert->copy(str, length, src_cs, dst_cs, &dummy_errors))
+    return TRUE;
+
+  return row->add_column(convert->ptr(), convert->length());
+}
+
+bool Protocol_local::store_tiny(longlong value)
+{
+  Ed_result_set *rs= m_result->get_cur_result_set();
+  Ed_row *row= rs->get_cur_row();
+
+  /* TODO: check medata type. */
+
+  char v= (char) value;
+
+  return row->add_column(&v, 1);
+}
+
+bool Protocol_local::store_short(longlong value)
+{
+  Ed_result_set *rs= m_result->get_cur_result_set();
+  Ed_row *row= rs->get_cur_row();
+
+  /* TODO: check medata type. */
+
+  int16 v= (int16) value;
+
+  return row->add_column(&v, 2);
+}
+
+bool Protocol_local::store_long(longlong value)
+{
+  Ed_result_set *rs= m_result->get_cur_result_set();
+  Ed_row *row= rs->get_cur_row();
+
+  /* TODO: check medata type. */
+
+  int32 v= (int32) value;
+
+  return row->add_column(&v, 4);
+}
+
+bool Protocol_local::store_longlong(longlong value, bool unsigned_flag)
+{
+  Ed_result_set *rs= m_result->get_cur_result_set();
+  Ed_row *row= rs->get_cur_row();
+
+  /* TODO: check medata type. */
+
+  int64 v= (int64) value;
+
+  return row->add_column(&v, 8);
+}
+
+bool Protocol_local::store_decimal(const my_decimal *value)
+{
+  Ed_result_set *rs= m_result->get_cur_result_set();
+  Ed_row *row= rs->get_cur_row();
+
+  /* TODO: check medata type. */
+
+  char buf[DECIMAL_MAX_STR_LENGTH];
+  String str(buf, sizeof (buf), &my_charset_bin);
+  my_decimal2string(E_DEC_FATAL_ERROR, value, 0, 0, 0, &str);
+
+  return row->add_column(str.ptr(), str.length());
+}
+
+bool Protocol_local::store(const char *str,
+                           size_t length,
+                           CHARSET_INFO *src_cs)
+{
+  CHARSET_INFO *dst_cs= this->thd->variables.character_set_results;
+  return store_string(str, length, src_cs, dst_cs);
+}
+
+bool Protocol_local::store(const char *str,
+                           size_t length,
+                           CHARSET_INFO *src_cs,
+                           CHARSET_INFO *dst_cs)
+{
+  return store_string(str, length, src_cs, dst_cs);
+}
+
+bool Protocol_local::store(MYSQL_TIME *time)
+{
+  Ed_result_set *rs= m_result->get_cur_result_set();
+  Ed_row *row= rs->get_cur_row();
+
+  /* TODO: check medata type. */
+
+  return row->add_column(time, sizeof (MYSQL_TIME));
+}
+
+bool Protocol_local::store_date(MYSQL_TIME *time)
+{
+  Ed_result_set *rs= m_result->get_cur_result_set();
+  Ed_row *row= rs->get_cur_row();
+
+  /* TODO: check medata type. */
+
+  return row->add_column(time, sizeof (MYSQL_TIME));
+}
+
+bool Protocol_local::store_time(MYSQL_TIME *time)
+{
+  Ed_result_set *rs= m_result->get_cur_result_set();
+  Ed_row *row= rs->get_cur_row();
+
+  /* TODO: check medata type. */
+
+  return row->add_column(time, sizeof (MYSQL_TIME));
+}
+
+bool Protocol_local::store(float value, uint32 decimals, String *buffer)
+{
+  Ed_result_set *rs= m_result->get_cur_result_set();
+  Ed_row *row= rs->get_cur_row();
+
+  /* TODO: check medata type. */
+
+  return row->add_column(&value, sizeof (float));
+}
+
+bool Protocol_local::store(double value, uint32 decimals, String *buffer)
+{
+  Ed_result_set *rs= m_result->get_cur_result_set();
+  Ed_row *row= rs->get_cur_row();
+
+  /* TODO: check medata type. */
+
+  return row->add_column(&value, sizeof (double));
+}
+
+bool Protocol_local::store(Field *field)
+{
+  if (field->is_null())
+    return store_null();
+  return field->send_binary(this);
+}
+
+bool Protocol_local::send_result_set_metadata(List<Item> *columns, uint)
+{
+  DBUG_ASSERT(m_result);
+
+  return m_result->add_result_set(columns);
+}
+
+bool Protocol_local::send_out_parameters(List<Item_param> *sp_params)
+{
+  return FALSE;
+}
+
+void Protocol_local::send_ok(uint server_status, uint statement_warn_count,
+                             ha_rows affected_rows, ulonglong last_insert_id,
+                             const char *message)
+{
+  m_result->send_ok(thd, server_status, statement_warn_count,
+                    affected_rows, last_insert_id, message);
+}
+
+void Protocol_local::send_eof(uint server_status, uint statement_warn_count)
+{
+  m_result->send_eof(thd, server_status, statement_warn_count);
+}
+
+void Protocol_local::send_error(uint sql_errno, const char *err_msg)
+{
+  m_result->send_error(thd, sql_errno, err_msg);
+}
+
+#ifdef EMBEDDED_LIBRARY
+void Protocol_local::remove_last_row()
+{ }
+#endif

=== modified file 'sql/protocol.h'
--- a/sql/protocol.h	2008-11-18 20:25:51 +0000
+++ b/sql/protocol.h	2008-11-27 18:31:59 +0000
@@ -17,6 +17,8 @@
 #pragma interface			/* gcc class implementation */
 #endif
 
+#include "sql_error.h"
+
 class i_string;
 class THD;
 class Item_param;
@@ -46,6 +48,15 @@ protected:
                       CHARSET_INFO *fromcs, CHARSET_INFO *tocs);
   bool store_string_aux(const char *from, size_t length,
                         CHARSET_INFO *fromcs, CHARSET_INFO *tocs);
+
+  virtual void send_ok(uint server_status, uint statement_warn_count,
+                       ha_rows affected_rows, ulonglong last_insert_id,
+                       const char *message);
+
+  virtual void send_eof(uint server_status, uint statement_warn_count);
+
+  virtual void send_error(uint sql_errno, const char *err_msg);
+
 public:
   Protocol() {}
   Protocol(THD *thd_arg) { init(thd_arg); }
@@ -106,23 +117,15 @@ public:
 #endif
   enum enum_protocol_type
   {
-    PROTOCOL_TEXT= 0, PROTOCOL_BINARY= 1
     /*
-      before adding here or change the values, consider that it is cast to a
-      bit in sql_cache.cc.
+      Before adding a new type, please make sure
+      there is enough storage for it in Query_cache_query_flags.
     */
+    PROTOCOL_TEXT= 0, PROTOCOL_BINARY= 1, PROTOCOL_LOCAL= 2
   };
   virtual enum enum_protocol_type type()= 0;
 
   void end_statement();
-
-  virtual void send_ok(uint server_status, uint statement_warn_count,
-                       ha_rows affected_rows, ulonglong last_insert_id,
-                       const char *message);
-
-  virtual void send_eof(uint server_status, uint statement_warn_count);
-
-  virtual void send_error(uint sql_errno, const char *err_msg);
 };
 
 
@@ -198,4 +201,267 @@ bool send_old_password_request(THD *thd)
 uchar *net_store_data(uchar *to,const uchar *from, size_t length);
 uchar *net_store_data(uchar *to,int32 from);
 uchar *net_store_data(uchar *to,longlong from);
+
+///////////////////////////////////////////////////////////////////////////
+
+/**
+  Protocol_local: a protocol to retrieve and store
+  a result set.
+*/
+
+class Ed_result;
+
+class Protocol_local :public Protocol
+{
+public:
+  inline Protocol_local(THD *thd, Ed_result *result)
+    :Protocol(thd), m_result(result)
+  {}
+
+public:
+  virtual void prepare_for_resend();
+  virtual bool write();
+  virtual bool store_null();
+  virtual bool store_tiny(longlong from);
+  virtual bool store_short(longlong from);
+  virtual bool store_long(longlong from);
+  virtual bool store_longlong(longlong from, bool unsigned_flag);
+  virtual bool store_decimal(const my_decimal *);
+  virtual bool store(const char *from, size_t length, CHARSET_INFO *cs);
+  virtual bool store(const char *from, size_t length,
+                     CHARSET_INFO *fromcs, CHARSET_INFO *tocs);
+  virtual bool store(MYSQL_TIME *time);
+  virtual bool store_date(MYSQL_TIME *time);
+  virtual bool store_time(MYSQL_TIME *time);
+  virtual bool store(float value, uint32 decimals, String *buffer);
+  virtual bool store(double value, uint32 decimals, String *buffer);
+  virtual bool store(Field *field);
+
+  virtual bool send_result_set_metadata(List<Item> *list, uint flags);
+  virtual bool send_out_parameters(List<Item_param> *sp_params);
+#ifdef EMBEDDED_LIBRARY
+  void remove_last_row();
+#endif
+  virtual enum enum_protocol_type type() { return PROTOCOL_LOCAL; };
+
+protected:
+  virtual void send_ok(uint server_status, uint statement_warn_count,
+                       ha_rows affected_rows, ulonglong last_insert_id,
+                       const char *message);
+
+  virtual void send_eof(uint server_status, uint statement_warn_count);
+
+  virtual void send_error(uint sql_errno, const char *err_msg);
+
+private:
+  bool store_string(const char *str, int length,
+                    CHARSET_INFO *src_cs, CHARSET_INFO *dst_cs);
+
+private:
+  Ed_result *m_result;
+};
+
+
+/**
+  Ed_column -- a class representing a column data in a row. Used with
+  Ed_row and Protocol_local.
+*/
+
+class Ed_column : public LEX_STRING, public Sql_alloc
+{
+public:
+  inline Ed_column()
+  {
+    str= NULL;
+    length= 0;
+  }
+
+  inline void set_data(MEM_ROOT *mem_root, const void *p_str, int p_length)
+  {
+    str= (char *) memdup_root(mem_root, p_str, p_length);
+    length= p_length;
+  }
+};
+
+///////////////////////////////////////////////////////////////////////////
+
+/**
+  Ed_result_set_metadata -- a class representing result set metadata. Used
+  with Ed_result_set and Protocol_local.
+*/
+
+class Ed_result_set_metadata : public Sql_alloc
+{
+public:
+  static Ed_result_set_metadata *create(MEM_ROOT *mem_root,
+                                        List<Item> *col_metadata);
+
+public:
+  inline int get_num_columns() const { return m_num_columns; }
+
+  inline const Send_field *get_column(int idx) const
+  { return &m_metadata[idx]; }
+
+private:
+  inline Ed_result_set_metadata() :
+    m_num_columns(0),
+    m_metadata(NULL)
+  { }
+
+private:
+  bool init(MEM_ROOT *mem_root, List<Item> *col_metadata);
+
+private:
+  int m_num_columns;
+  Send_field *m_metadata;
+};
+
+///////////////////////////////////////////////////////////////////////////
+
+/**
+  Ed_row -- a class representing a row in a result set. Used with
+  Ed_result_set and Protocol_local.
+*/
+class Ed_row : public Sql_alloc
+{
+public:
+  static Ed_row *create(MEM_ROOT *mem_root,
+                        const Ed_result_set_metadata *metadata);
+
+public:
+  bool add_null();
+  bool add_column(const void *data_ptr, int data_length);
+
+public:
+  inline const Ed_result_set_metadata *get_metadata() const
+  { return m_metadata; }
+
+  inline const Ed_column *get_column(int idx) const
+  { return &m_columns[idx]; }
+
+  inline const Ed_column &operator [](int idx) const
+  { return *get_column(idx); }
+
+  inline int get_current_column_index() const
+  { return m_current_column_index; }
+
+private:
+  inline Ed_row(MEM_ROOT *mem_root,
+                const Ed_result_set_metadata *metadata) :
+    m_mem_root(mem_root),
+    m_metadata(metadata),
+    m_current_column_index(0)
+  { }
+
+  bool init();
+
+private:
+  MEM_ROOT *m_mem_root;
+  const Ed_result_set_metadata *m_metadata;
+  Ed_column *m_columns;
+  int m_current_column_index;
+};
+
+///////////////////////////////////////////////////////////////////////////
+
+/**
+  Ed_result_set -- a class representing one result set. Used with Ed_result
+  and Protocol_local.
+*/
+class Ed_result_set : public Sql_alloc
+{
+public:
+  static Ed_result_set *create(MEM_ROOT *mem_root,
+                               List<Item> *col_metadata);
+
+private:
+  inline Ed_result_set(MEM_ROOT *mem_root) :
+    m_mem_root(mem_root),
+    m_metadata(NULL),
+    m_current_row(NULL)
+  { }
+
+private:
+  bool init(List<Item> *col_metadata);
+
+public:
+  inline const Ed_result_set_metadata *get_metadata() const
+  { return m_metadata; }
+
+  inline List<Ed_row> *data()
+  { return &m_data; }
+
+  inline Ed_row *get_cur_row()
+  { return m_current_row; }
+
+  Ed_row *add_row();
+
+private:
+  MEM_ROOT *m_mem_root;
+
+  Ed_result_set_metadata *m_metadata;
+  List<Ed_row> m_data;
+
+  Ed_row *m_current_row;
+};
+
+///////////////////////////////////////////////////////////////////////////
+
+/*
+  Ed_result -- a class representing results for an SQL statement execution.
+  Used with Protocol_local.
+*/
+class Ed_result : public List<Ed_result_set>
+{
+public:
+  Ed_result();
+  ~Ed_result();
+
+public:
+  inline Ed_result_set *get_cur_result_set()
+  { return m_current_result_set; }
+
+  bool add_result_set(List<Item> *col_metadata);
+
+public:
+  void send_ok(THD *thd, uint server_status, uint statement_warn_count,
+               ha_rows affected_rows, ulonglong last_insert_id,
+               const char *message);
+
+  void send_eof(THD *thd, uint server_status, uint statement_warn_count);
+
+  void send_error(THD *thd, uint sql_errno, const char *err_msg);
+
+  void begin_statement(THD *thd);
+  void end_statement(THD *thd);
+
+public:
+  inline uint get_status() const { return m_status; }
+  inline uint get_server_status() const { return m_server_status; }
+  inline ha_rows get_affected_rows() const { return m_affected_rows; }
+  inline ulonglong get_last_insert_id() const { return m_last_insert_id; }
+  inline uint get_sql_errno() const { return m_sql_errno; }
+  inline const char *get_message() const { return m_message; }
+
+  inline uint get_statement_warn_count() const
+  { return m_warning_info.statement_warn_count(); }
+
+  inline List<MYSQL_ERROR> &get_warnings()
+  { return m_warning_info.warn_list(); }
+
+private:
+  MEM_ROOT m_mem_root;
+  Ed_result_set *m_current_result_set;
+
+private:
+  uint m_status;
+  uint m_server_status;
+  ha_rows m_affected_rows;
+  ulonglong m_last_insert_id;
+  uint m_sql_errno;
+  char m_message[MYSQL_ERRMSG_SIZE];
+
+  Warning_info m_warning_info;
+  Warning_info *m_warning_info_saved;
+};
 

=== modified file 'sql/share/errmsg.txt'
--- a/sql/share/errmsg.txt	2008-11-07 17:07:58 +0000
+++ b/sql/share/errmsg.txt	2008-11-21 19:05:10 +0000
@@ -6397,7 +6397,7 @@ ER_BACKUP_GET_META_PRIV
 ER_BACKUP_CANT_RESTORE_PRIV
         eng "Could not execute grant '%-.64s'."
 ER_BACKUP_GRANT_SKIPPED
-        eng "The grant '%-.64s' was skipped because the user does not exist."
+        eng "The grant '%-.64s' for the user %-.64s was skipped because the user does not exist."
 ER_BACKUP_GRANT_WRONG_DB
         eng "The grant '%-.64s' failed. Database not included in the backup image."
 ER_BACKUP_LOGPATHS

=== modified file 'sql/si_objects.cc'
--- a/sql/si_objects.cc	2008-10-27 13:06:21 +0000
+++ b/sql/si_objects.cc	2008-12-02 11:27:23 +0000
@@ -1,2081 +1,1741 @@
-/**
-   @file
+/* Copyright (C) 2008 MySQL AB
 
-   This file defines the API for the following object services:
-     - serialize database objects into a string;
-     - materialize (deserialize) object from a string;
-     - enumerating objects;
-     - finding dependencies for objects;
-     - executor for SQL statements;
-     - wrappers for controlling the DDL Blocker;
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; version 2 of the License.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307 USA */
 
-  The methods defined below are used to provide server functionality to
-  and permitting an isolation layer for the client (caller).
-*/
+/** @file Server Service Interface for Backup: the implementation.  */
 
 #include "mysql_priv.h"
+
 #include "si_objects.h"
 #include "ddl_blocker.h"
 #include "sql_show.h"
+#include "sql_trigger.h"
+#include "sp.h"
+#include "sp_head.h" // for sp_add_to_query_tables().
+
 #ifdef HAVE_EVENT_SCHEDULER
 #include "events.h"
 #include "event_data_objects.h"
 #include "event_db_repository.h"
 #endif
-#include "sql_trigger.h"
-#include "sp.h"
-#include "sp_head.h" // for sp_add_to_query_tables().
-
-TABLE *create_schema_table(THD *thd, TABLE_LIST *table_list); // defined in sql_show.cc
 
 DDL_blocker_class *DDL_blocker= NULL;
 
 ///////////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////
+
+#define STR(x) (int) (x).length(), (x).ptr()
+
+#define LXS_INIT(x) {((char *) (x)), ((size_t) (sizeof (x) - 1))}
+
+///////////////////////////////////////////////////////////////////////////
 
 namespace {
 
-// Helper methods
+///////////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////
 
 /**
-  Execute the SQL string passed.
-
-  This is a private helper function to the implementation.
+  Si_session_context defines a way to save/reset/restore session context
+  for SI-operations.
 */
-int silent_exec(THD *thd, String *query)
-{
-  Vio *save_vio= thd->net.vio;
 
-  DBUG_PRINT("si_objects",("executing %s",query->c_ptr()));
+class Si_session_context
+{
+public:
+  inline Si_session_context()
+  { }
 
-  /*
-    Note: the change net.vio idea taken from execute_init_command in
-    sql_parse.cc
-   */
-  thd->net.vio= 0;
-
-  thd->query=         query->c_ptr();
-  thd->query_length=  query->length();
-
-  thd->set_time();
-  pthread_mutex_lock(&::LOCK_thread_count);
-  thd->query_id= ::next_query_id();
-  pthread_mutex_unlock(&::LOCK_thread_count);
+public:
+  void save_si_ctx(THD *thd);
+  void reset_si_ctx(THD *thd);
+  void restore_si_ctx(THD *thd);
 
-  /*
-    @todo The following is a work around for online backup and the DDL blocker.
-          It should be removed when the generalized solution is in place.
-          This is needed to ensure the restore (which uses DDL) is not blocked
-          when the DDL blocker is engaged.
-  */
-  thd->DDL_exception= TRUE;
+private:
+  ulong m_sql_mode_saved;
+  CHARSET_INFO *m_client_cs_saved;
+  CHARSET_INFO *m_results_cs_saved;
+  CHARSET_INFO *m_connection_cl_saved;
+  Time_zone *m_tz_saved;
+  TABLE *m_tmp_tables_saved;
 
-  /*
-    Note: This is a copy and paste from the code in sql_parse.cc.
-          See "case COM_QUERY:".
-  */
-  const char *found_semicolon= thd->query;
-  char *packet_end= thd->query + thd->query_length;
-  mysql_parse(thd, thd->query, thd->query_length, &found_semicolon);
-  while (!thd->killed && found_semicolon && !thd->is_error())
-  {
-    char *next_packet= (char*) found_semicolon;
-    /*
-      Multiple queries exits, execute them individually
-    */
-    close_thread_tables(thd);
-    ulong length= (ulong)(packet_end - next_packet);
+private:
+  Si_session_context(const Si_session_context &);
+  Si_session_context &operator =(const Si_session_context &);
+};
 
-    /* Remove garbage at start of query */
-    while (my_isspace(thd->charset(), *next_packet) && length > 0)
-    {
-      next_packet++;
-      length--;
-    }
-    pthread_mutex_lock(&LOCK_thread_count);
-    thd->query_length= length;
-    thd->query= next_packet;
-    thd->query_id= next_query_id();
-    thd->set_time(); /* Reset the query start time. */
-    /* TODO: set thd->lex->sql_command to SQLCOM_END here */
-    pthread_mutex_unlock(&LOCK_thread_count);
-    mysql_parse(thd, next_packet, length, & found_semicolon);
-  }
+///////////////////////////////////////////////////////////////////////////
 
-  close_thread_tables(thd);
+/**
+  Preserve the following session attributes:
+    - sql_mode;
+    - character_set_client;
+    - character_set_results;
+    - collation_connection;
+    - time_zone;
 
-  thd->net.vio= save_vio;
+  Remember also session temporary tables.
+*/
 
-  if (thd->is_error())
-  {
-    DBUG_PRINT("restore",
-              ("error executing query %s!", thd->query));
-    DBUG_PRINT("restore",("last error (%d): %s",thd->net.last_errno
-                                               ,thd->net.last_error));
-    return thd->net.last_errno ? (int)thd->net.last_errno : -1;
-  }
+void Si_session_context::save_si_ctx(THD *thd)
+{
+  DBUG_ENTER("Si_session_context::save_si_ctx");
+  m_sql_mode_saved= thd->variables.sql_mode;
+  m_client_cs_saved= thd->variables.character_set_client;
+  m_results_cs_saved= thd->variables.character_set_results;
+  m_connection_cl_saved= thd->variables.collation_connection;
+  m_tz_saved= thd->variables.time_zone;
+  m_tmp_tables_saved= thd->temporary_tables;
 
-  return 0;
+  DBUG_VOID_RETURN;
 }
 
-/*
-  This method gets the create statement for a procedure or function.
+///////////////////////////////////////////////////////////////////////////
+
+/**
+  Reset session state to the following:
+    - sql_mode: 0
+    - character_set_client: utf8
+    - character_set_results: binary (to fetch results w/o conversion)
+    - collation_connection: utf8
+
+  Temporary tables should be ignored while looking for table structures.
+  We want to deal with real tables, not temporary ones.
 */
-bool serialize_routine(THD *thd,
-                     int type,
-                      String db_name,
-                      String r_name,
-                      String *string)
+
+void Si_session_context::reset_si_ctx(THD *thd)
 {
-  int ret= 0;
-  sp_head *sp;
-  sp_name *routine_name;
-  LEX_STRING sql_mode;
-  DBUG_ENTER("serialize_routine");
-  DBUG_PRINT("serialize_routine", ("name: %s@%s", db_name.c_ptr(),
-             r_name.c_ptr()));
-
-  DBUG_ASSERT(type == TYPE_ENUM_PROCEDURE || type == TYPE_ENUM_FUNCTION);
-  sp_cache **cache = type == TYPE_ENUM_PROCEDURE ?
-                     &thd->sp_proc_cache : &thd->sp_func_cache;
-  LEX_STRING db;
-  db.str= db_name.c_ptr();
-  db.length= db_name.length();
-  LEX_STRING name;
-  name.str= r_name.c_ptr();
-  name.length= r_name.length();
-  routine_name= new sp_name(db, name, true);
-  routine_name->init_qname(thd);
-  if (type == TYPE_ENUM_PROCEDURE)
-    thd->variables.max_sp_recursion_depth++;
-  if ((sp= sp_find_routine(thd, type, routine_name, cache, FALSE)))
-  {
-    sys_var_thd_sql_mode::symbolic_mode_representation(thd,
-      sp->m_sql_mode, &sql_mode);
-    Stored_program_creation_ctx *sp_ctx= sp->get_creation_ctx();
+  DBUG_ENTER("Si_session_context::reset_si_ctx");
 
-    /*
-      Prepend sql_mode command.
-    */
-    string->append("SET SQL_MODE = '");
-    string->append(sql_mode.str);
-    string->append("'; ");
+  thd->variables.sql_mode= 0;
 
-    /*
-      append character set client charset information
-    */
-    string->append("SET CHARACTER_SET_CLIENT = '");
-    string->append(sp_ctx->get_client_cs()->csname);
-    string->append("'; ");
+  thd->variables.character_set_client= system_charset_info;
+  thd->variables.character_set_results= &my_charset_bin;
+  thd->variables.collation_connection= system_charset_info;
+  thd->update_charset();
 
-    /*
-      append collation_connection information
-    */
-    string->append("SET COLLATION_CONNECTION = '");
-    string->append(sp_ctx->get_connection_cl()->name);
-    string->append("'; ");
-
-    /*
-      append collation_connection information
-    */
-    string->append("SET COLLATION_DATABASE = '");
-    string->append(sp_ctx->get_db_cl()->name);
-    string->append("'; ");
+  thd->temporary_tables= NULL;
 
-    string->append(sp->m_defstr.str);
-  }
-  else
-  {
-    string->length(0);
-    ret= TRUE;
-  }
-  if (type == TYPE_ENUM_PROCEDURE)
-    thd->variables.max_sp_recursion_depth--;
-  DBUG_RETURN(ret);
+  DBUG_VOID_RETURN;
 }
 
-/*
-  This method calls silent_exec() while saving the context
-  information before the call and restoring after the call.
+///////////////////////////////////////////////////////////////////////////
 
-  If save_timezone, it also saves and restores the timezone.
+/**
+  Restore session state.
 */
-bool execute_with_ctx(THD *thd, String *query, bool save_timezone)
-{
-  bool ret= false;
-  ulong orig_sql_mode;
-  CHARSET_INFO *orig_char_set_client;
-  CHARSET_INFO *orig_coll_conn;
-  CHARSET_INFO *orig_coll_db;
-  Time_zone *tm_zone;
-  DBUG_ENTER("Obj::execute_with_ctx()");
-
-  /*
-    Preserve SQL_MODE, CHARACTER_SET_CLIENT, COLLATION_CONNECTION,
-    and COLLATION_DATABASE.
-  */
-  orig_sql_mode= thd->variables.sql_mode;
-  orig_char_set_client= thd->variables.character_set_client;
-  orig_coll_conn= thd->variables.collation_connection;
-  orig_coll_db= thd->variables.collation_database;
 
-  /*
-    Preserve timezone.
-  */
-  if (save_timezone)
-    tm_zone= thd->variables.time_zone;
+void Si_session_context::restore_si_ctx(THD *thd)
+{
+  DBUG_ENTER("Si_session_context::restore_si_ctx");
 
-  ret= silent_exec(thd, query);
+  thd->variables.sql_mode= m_sql_mode_saved;
+  thd->variables.time_zone= m_tz_saved;
 
-  /*
-    Restore SQL_MODE, CHARACTER_SET_CLIENT, COLLATION_CONNECTION,
-    and COLLATION_DATABASE.
-  */
-  thd->variables.sql_mode= orig_sql_mode;
-  thd->variables.character_set_client= orig_char_set_client;
-  thd->variables.collation_connection= orig_coll_conn;
-  thd->variables.collation_database= orig_coll_db;
+  thd->variables.collation_connection= m_connection_cl_saved;
+  thd->variables.character_set_results= m_results_cs_saved;
+  thd->variables.character_set_client= m_client_cs_saved;
+  thd->update_charset();
 
-  /*
-    Restore timezone.
-  */
-  if (save_timezone)
-    thd->variables.time_zone= tm_zone;
+  thd->temporary_tables= m_tmp_tables_saved;
 
-  DBUG_RETURN(ret);
+  DBUG_VOID_RETURN;
 }
 
-/*
-  Drops an object.
-
-  obj_name is the name of the object e.g., DATABASE, PROCEDURE, etc.
-  name1 is the db name  (blank for database objects)
-  name2 is the name of the object
-*/
-bool drop_object(THD *thd, const char *obj_name, String *name1, String *name2)
-{
-  DBUG_ENTER("Obj::drop_object()");
-  String cmd;
-  cmd.length(0);
-  cmd.append("DROP ");
-  cmd.append(obj_name);
-  cmd.append(" IF EXISTS ");
-  if (name1 && (name1->length() > 0))
-  {
-    append_identifier(thd, &cmd, name1->c_ptr(), name1->length());
-    cmd.append(".");
-  }
-  append_identifier(thd, &cmd, name2->c_ptr(), name2->length());
-  DBUG_RETURN(silent_exec(thd, &cmd));
-}
+///////////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////
 
 /**
-  Open given table in @c INFORMATION_SCHEMA database.
-
-  This is a private helper function to the implementation.
+  Execute one DML statement in a backup-specific context. Result set and
+  warning information are stored in the output parameter. Some session
+  attributes are preserved and reset to predefined values before query
+  execution (@see Si_session_context).
 
-  @param[in] thd  Thread context
-  @param[in] st   Schema table enum
-  @param[in] db_list List of databases for select condition
+  @param[in]  thd         Thread context.
+  @param[in]  query       SQL query to be executed.
+  @param[out] ed_result   A place to store result and warnings.
 
-  @note: The select condition is designed to form a WHERE clause based on
-  the database/schema column of the information_schema views. Most views have
-  a database/schema column but for those that do not, you must ignore the
-  selection condition by passing db_list = NULL.
-
-  @retval TABLE* The schema table
+  @return Error status.
+    @retval TRUE on error.
+    @retval FALSE on success.
 */
-TABLE* open_schema_table(THD *thd, ST_SCHEMA_TABLE *st, List<LEX_STRING> *db_list)
-{
-  TABLE *t;
-  TABLE_LIST arg;
-  my_bitmap_map *old_map;
-
-  bzero( &arg, sizeof(TABLE_LIST) );
 
-  // set context for create_schema_table call
-  arg.schema_table= st;
-  arg.alias=        NULL;
-  arg.select_lex=   NULL;
+bool
+run_service_interface_sql(THD *thd, const LEX_STRING *query,
+                          Ed_result *ed_result)
+{
+  Si_session_context session_context;
 
-  t= create_schema_table(thd,&arg); // Note: callers must free t.
+  DBUG_ENTER("run_service_interface_sql");
+  DBUG_PRINT("run_service_interface_sql",
+             ("query: %.*s",
+              (int) query->length, (const char *) query->str));
 
-  if( !t ) return NULL; // error!
+  session_context.save_si_ctx(thd);
+  session_context.reset_si_ctx(thd);
 
-  /*
-   Temporarily set thd->lex->wild to NULL to keep st->fill_table
-   happy.
-  */
-  ::String *wild= thd->lex->wild;
-  ::enum_sql_command command= thd->lex->sql_command;
+  bool rc= mysql_execute_direct(thd, *query, ed_result);
 
-  thd->lex->wild = NULL;
-  thd->lex->sql_command = enum_sql_command(0);
+  session_context.restore_si_ctx(thd);
 
-  // context for fill_table
-  arg.table= t;
+  DBUG_RETURN(rc);
+}
 
-  old_map= tmp_use_all_columns(t, t->read_set);
+///////////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////
 
-  /*
-    Create a selection condition only if db_list is defined.
-  */
-  if (db_list)
-    st->fill_table(thd, &arg, obs::create_db_select_condition(thd, t, db_list));
-  else
-    st->fill_table(thd, &arg, NULL);
+/**
+  Update THD with the warnings from the given list.
 
-  tmp_restore_column_map(t->read_set, old_map);
+  @param[in]  thd   Thread context.
+  @parampin]  src   Warning list.
+*/
 
-  // undo changes to thd->lex
-  thd->lex->wild= wild;
-  thd->lex->sql_command= command;
+void copy_warnings(THD *thd, List<MYSQL_ERROR> *src)
+{
+  List_iterator_fast<MYSQL_ERROR> err_it(*src);
+  MYSQL_ERROR *err;
 
-  return t;
+  while ((err= err_it++))
+    push_warning(thd, err->level, err->code, err->msg);
 }
 
-/*
-  Prepend the USE DB <obj> command.
+///////////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////
+
+/**
+  Table_name_key defines a hash key, which includes database and tables
+  names.
 */
-void prepend_db(THD *thd, String *serialization, String *db_name)
+
+struct Table_name_key
 {
-  DBUG_ENTER("Obj::prepend_db()");
-  /*
-    prepend "USE db" statement
-  */
-  serialization->length(0);
-  serialization->append("USE ");
-  append_identifier(thd, serialization, db_name->c_ptr(), db_name->length());
-  serialization->append("; ");
-  DBUG_VOID_RETURN;
-}
+public:
+  LEX_STRING db_name;
+  LEX_STRING table_name;
+  LEX_STRING key;
+
+public:
+  void init_from_mdl_key(const char *key_arg, size_t key_length_arg,
+                         char *key_buff_arg);
+};
 
 ///////////////////////////////////////////////////////////////////////////
 
-struct Table_name_key
+void
+Table_name_key::init_from_mdl_key(const char *key_arg, size_t key_length_arg,
+                                  char *key_buff_arg)
 {
-  Table_name_key(const char *db_name_str,
-                 uint db_name_length,
-                 const char *table_name_str,
-                 uint table_name_length)
-  {
-    db_name.copy(db_name_str, db_name_length, system_charset_info);
-    table_name.copy(table_name_str, table_name_length, system_charset_info);
+  key.str= key_buff_arg;
+  key.length= key_length_arg - 4; /* Skip MDL object type */
+  memcpy(key.str, key_arg + 4, key.length);
 
-    key.length(0);
-    key.append(db_name);
-    key.append(".");
-    key.append(table_name);
-  }
+  db_name.str= key.str;
+  db_name.length= strlen(db_name.str);
+  table_name.str= db_name.str + db_name.length + 1; /* Skip \0 */
+  table_name.length= strlen(table_name.str);
+}
 
-  String db_name;
-  String table_name;
+///////////////////////////////////////////////////////////////////////////
 
-  String key;
-};
+extern "C" {
 
-uchar *
+static uchar *
 get_table_name_key(const uchar *record,
                    size_t *key_length,
                    my_bool not_used __attribute__((unused)))
 {
-  Table_name_key *tnk= (Table_name_key *) record;
-  *key_length= tnk->key.length();
-  return (uchar *) tnk->key.c_ptr_safe();
+  Table_name_key *table_name_key= (Table_name_key *) record;
+  *key_length= table_name_key->key.length;
+  return (uchar *) table_name_key->key.str;
 }
 
-void delete_table_name_key(void *data)
+///////////////////////////////////////////////////////////////////////////
+
+static void
+free_table_name_key(void *data)
 {
-  Table_name_key *tnk= (Table_name_key *) data;
-  delete tnk;
+  Table_name_key *table_name_key= (Table_name_key *) data;
+  my_free(table_name_key, MYF(0));
 }
 
-}
+} // end of extern "C"
 
 ///////////////////////////////////////////////////////////////////////////
-
-namespace obs {
+///////////////////////////////////////////////////////////////////////////
 
 /**
-  Build a where clause for list of databases.
-
-  This method is used to help improve the efficiency of queries against
-  information schema tables. It builds a condition tree of the form
-  db_col IN ('a','b','c') where a,b,c are database names.
-
-  @param[in] thd      Thread context.
-  @param[in] t        The table to operate on.
-  @param[in] db_list  The list of databases in form List<LEX_STRING>
-
-  @returns NULL if no databases in list or pointer to COND tree.
-*/
-COND *create_db_select_condition(THD *thd,
-                                 TABLE *t,
-                                 List<LEX_STRING> *db_list)
-{
-  List<Item> in_db_list;
-  List_iterator< ::LEX_STRING> it(*db_list);
-  ::LEX_STRING *db;
-  DBUG_ENTER("Obj::create_select_condition()");
-
-  /*
-    If no list of databases, just return NULL
-  */
-  if (!db_list->elements)
-    DBUG_RETURN(NULL);
+  Int_value is a wrapper for unsigned long type to be used with
+  the String_stream class.
+*/
 
-  /*
-    Build an inclusion list in the form of 'a', 'b', etc.
-  */
-  while ((db= it++))
-  {
-    Item *db_name= new Item_string(db->str, db->length, system_charset_info);
-    db_name->fix_fields(thd, &db_name);
-    db_name->next= NULL;
-    in_db_list.push_front(db_name);
-  }
+struct Int_value
+{
+  unsigned long m_value;
 
-  /*
-    Build the compared field "table_schema" and link to temp table field.
-  */
-  Item *db_field= new Item_field(thd, thd->lex->current_context(), t->field[1]);
-  in_db_list.push_front(db_field);
+  Int_value(unsigned long value_arg)
+    :m_value(value_arg)
+  { }
+};
 
-  /*
-    Build the in function item comparison and add list of databases.
-  */
-  Item_func_in *in_cond = new Item_func_in(in_db_list);
-  in_cond->fix_fields(thd, (Item **)&in_cond);
-  in_cond->fix_length_and_dec();
+///////////////////////////////////////////////////////////////////////////
 
-  DBUG_RETURN(in_cond);
-}
+/*
+  C_str is a wrapper for C-string (const char *) to be used with the
+  String_stream class.
+*/
 
-///////////////////////////////////////////////////////////////////////////
+struct C_str
+{
+  LEX_STRING lex_string;
 
-//
-// Implementation: object impl classes.
-//
+  inline C_str(const char *str, size_t length)
+  {
+    lex_string.str= (char *) str;
+    lex_string.length= length;
+  }
+};
 
 ///////////////////////////////////////////////////////////////////////////
 
 /**
-   @class DatabaseObj
+  @class String_stream
 
-   This class provides an abstraction to a database object for creation and
-   capture of the creation data.
+  This class provides a convenient way to create a dynamic string from
+  C-strings in different forms (LEX_STRING, const char *) and integer
+  constants.
 */
-class DatabaseObj : public Obj
+
+class String_stream
 {
 public:
-  DatabaseObj(const String *db_name);
+  inline String_stream()
+    : m_buffer(&m_container)
+  { }
 
-public:
-  virtual bool materialize(uint serialization_version,
-                           const String *serialization);
+  inline String_stream(String *dst)
+    : m_buffer(dst)
+  { }
 
-  const String* get_name()
-  { return &m_db_name; }
+public:
+  inline String *str() { return m_buffer; }
 
-  const String *get_db_name()
+  inline const LEX_STRING *lex_string()
   {
-    return &m_db_name;
+    m_lex_string= m_buffer->lex_string();
+    return &m_lex_string;
   }
 
-private:
-  // These attributes are to be used only for serialization.
-  String m_db_name;
-
-  bool drop(THD *thd);
-  virtual bool do_serialize(THD *thd, String *serialization);
-  virtual bool do_execute(THD *thd);
+  inline void reset() { m_buffer->length(0); }
 
-private:
-  // These attributes are to be used only for materialization.
-  String m_create_stmt;
-};
-
-///////////////////////////////////////////////////////////////////////////
-
-/**
-   @class TableObj
-
-   This class provides an abstraction to a table object for creation and
-   capture of the creation data.
-*/
-class TableObj : public Obj
-{
 public:
-  TableObj(const String *db_name,
-           const String *table_name,
-           bool table_is_view);
+  String_stream &operator <<(const Int_value &v);
 
-public:
-  virtual bool materialize(uint serialization_version,
-                           const String *serialization);
+  inline String_stream &operator <<(const C_str &v)
+  {
+    m_buffer->append(v.lex_string.str, v.lex_string.length);
+    return *this;
+  }
+
+  inline String_stream &operator <<(const LEX_STRING *v)
+  {
+    m_buffer->append(v->str, v->length);
+    return *this;
+  }
 
-  const String* get_name()
-  { return &m_table_name; }
+  inline String_stream &operator <<(const String *v)
+  {
+    m_buffer->append(v->ptr(), v->length());
+    return *this;
+  }
 
-  const String *get_db_name()
+  String_stream &operator <<(const char *v)
   {
-    return &m_db_name;
+    m_buffer->append(v);
+    return *this;
   }
 
 private:
-  // These attributes are to be used only for serialization.
-  String m_db_name;
-  String m_table_name;
-  bool m_table_is_view;
+  String m_container;
+  String *m_buffer;
+  LEX_STRING m_lex_string;
+};
 
-  bool drop(THD *thd);
-  virtual bool do_serialize(THD *thd, String *serialization);
-  virtual bool do_execute(THD *thd);
+///////////////////////////////////////////////////////////////////////////
 
-private:
-  // These attributes are to be used only for materialization.
-  String m_create_stmt;
+String_stream &String_stream::operator <<(const Int_value &int_value)
+{
+  char buffer[13];
+  my_snprintf(buffer, sizeof (buffer), "%lu", int_value.m_value);
 
-private:
-  bool serialize_table(THD *thd, String *serialization);
-  bool serialize_view(THD *thd, String *serialization);
-};
+  m_buffer->append(buffer);
+  return *this;
+}
 
 ///////////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////
 
 /**
-  @class TriggerObj
+  @class Out_stream
 
-  This class provides an abstraction to a trigger object for creation and
-  capture of the creation data.
+  This class encapsulates the semantic of creating the serialization image.
+  The image is actually a list of strings. Strings may contain any data
+  (including binary and new-line characters). String is length-coded, which
+  means there is string length before the string data.
+
+  The format is as follows:
+    <string length> <space> <string data> \n
+
+  Example:
+    12 Hello,
+    world
+    5 qwerty
 */
-class TriggerObj : public Obj
+
+class Out_stream
 {
 public:
-  TriggerObj(const String *db_name,
-             const String *trigger_name);
+  inline Out_stream(String *image) :
+    m_image(image)
+  { }
 
 public:
-  virtual bool materialize(uint serialization_version,
-                           const String *serialization);
+  Out_stream &operator <<(const LEX_STRING *query);
 
-  const String* get_name()
-  { return &m_trigger_name; }
-
-  const String *get_db_name()
+  inline Out_stream &operator <<(const char *query)
   {
-    return &m_db_name;
+    LEX_STRING str= { (char *) query, strlen(query) };
+    return Out_stream::operator <<(&str);
   }
 
-private:
-  // These attributes are to be used only for serialization.
-  String m_db_name;
-  String m_trigger_name;
+  inline Out_stream &operator <<(const String *query)
+  {
+    LEX_STRING str= { (char *) query->ptr(), query->length() };
+    return Out_stream::operator <<(&str);
+  }
 
-  bool drop(THD *thd);
-  virtual bool do_serialize(THD *thd, String *serialization);
-  virtual bool do_execute(THD *thd);
+  inline Out_stream &operator <<(String_stream &s_stream)
+  {
+    return Out_stream::operator <<(s_stream.str());
+  }
 
 private:
-  // These attributes are to be used only for materialization.
-  String m_create_stmt;
+  String *m_image;
 };
 
 ///////////////////////////////////////////////////////////////////////////
 
+Out_stream &Out_stream::operator <<(const LEX_STRING *query)
+{
+  String_stream s_stream(m_image);
+
+  s_stream <<
+    Int_value(query->length) << " " << query << "\n";
+
+  return *this;
+}
+
+///////////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////
+
 /**
-  @class StoredProcObj
+  @class In_stream
 
-  This class provides an abstraction to a stored procedure object for creation
-  and capture of the creation data.
+  This class encapsulates the semantic of reading from the serialization image.
+  For the format definition of the serialization image, @see Out_stream.
 */
-class StoredProcObj : public Obj
+
+class In_stream
 {
 public:
-  StoredProcObj(const String *db_name,
-                const String *stored_proc_name);
+  inline In_stream(uint image_version, const String *image)
+    : m_image_version(image_version),
+      m_image(image),
+      m_read_ptr(m_image->ptr()),
+      m_end_ptr(m_image->ptr() + m_image->length())
+  { }
 
 public:
-  virtual bool materialize(uint serialization_version,
-                           const String *serialization);
-
-  const String* get_name()
-  { return &m_stored_proc_name; }
-
-  const String *get_db_name()
-  {
-    return &m_db_name;
-  }
+  bool next(LEX_STRING *chunk);
 
 private:
-  // These attributes are to be used only for serialization.
-  String m_db_name;
-  String m_stored_proc_name;
-
-  bool drop(THD *thd);
-  virtual bool do_serialize(THD *thd, String *serialization);
-  virtual bool do_execute(THD *thd);
-
-private:
-  // These attributes are to be used only for materialization.
-  String m_create_stmt;
+  uint m_image_version;
+  const String *m_image;
+  const char *m_read_ptr;
+  const char *m_end_ptr;
 };
 
 ///////////////////////////////////////////////////////////////////////////
 
-/**
-  @class StoredFuncObj
-
-  This class provides an abstraction to a stored function object for creation
-  and capture of the creation data.
-*/
-class StoredFuncObj : public Obj
+bool In_stream::next(LEX_STRING *chunk)
 {
-public:
-  StoredFuncObj(const String *db_name,
-                const String *stored_func_name);
-
-public:
-  virtual bool materialize(uint serialization_version,
-                           const String *serialization);
+  if (m_read_ptr >= m_end_ptr)
+    return TRUE;
 
-  const String* get_name()
-  { return &m_stored_func_name; }
+  const char *delimiter_ptr=
+    my_strchr(system_charset_info, m_read_ptr, m_end_ptr, ' ');
 
-  const String *get_db_name()
+  if (!delimiter_ptr)
   {
-    return &m_db_name;
+    m_read_ptr= m_end_ptr;
+    return TRUE;
   }
 
-private:
-  // These attributes are to be used only for serialization.
-  String m_db_name;
-  String m_stored_func_name;
+  char buffer[STRING_BUFFER_USUAL_SIZE];
+  int n= delimiter_ptr - m_read_ptr;
 
-  bool drop(THD *thd);
-  virtual bool do_serialize(THD *thd, String *serialization);
-  virtual bool do_execute(THD *thd);
+  memcpy(buffer, m_read_ptr, n);
+  buffer[n]= 0;
 
-private:
-  // These attributes are to be used only for materialization.
-  String m_create_stmt;
-};
+  chunk->str= (char *) delimiter_ptr + 1;
+  chunk->length= atoi(buffer);
 
-///////////////////////////////////////////////////////////////////////////
-#ifdef HAVE_EVENT_SCHEDULER
-/**
-  @class EventObj
+  m_read_ptr+= n    /* chunk length */
+    + 1             /* delimiter (a space) */
+    + chunk->length /* chunk */
+    + 1;            /* chunk delimiter (\n) */
 
-  This class provides an abstraction to a event object for creation and capture
-  of the creation data.
-*/
-class EventObj : public Obj
-{
-public:
-  EventObj(const String *db_name,
-           const String *event_name);
+  return FALSE;
+}
 
-public:
-  virtual bool materialize(uint serialization_version,
-                           const String *serialization);
+///////////////////////////////////////////////////////////////////////////
 
-  const String* get_name()
-  { return &m_event_name; }
+} // end of anonymous namespace
 
-  const String *get_db_name()
-  {
-    return &m_db_name;
-  }
-
-private:
-  // These attributes are to be used only for serialization.
-  String m_db_name;
-  String m_event_name;
+///////////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////
 
-  bool drop(THD *thd);
-  virtual bool do_serialize(THD *thd, String *serialization);
-  virtual bool do_execute(THD *thd);
+namespace obs {
 
-private:
-  // These attributes are to be used only for materialization.
-  String m_create_stmt;
-};
-#endif  // HAVE_EVENT_SCHEDULER
+///////////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////
 
 /**
-   @class TablespaceObj
+  @class Abstract_obj
 
-   This class provides an abstraction to a user object for creation and
-   capture of the creation data.
+  This class is a base class for all other Obj implementations.
 */
-class TablespaceObj : public Obj
+
+class Abstract_obj : public Obj
 {
 public:
-  TablespaceObj(const String *ts_name);
+  virtual inline const String *get_name() const    { return &m_id; }
+  virtual inline const String *get_db_name() const { return NULL; }
 
 public:
-  virtual bool do_serialize(THD *thd, String *serialization);
+  /**
+    Serialize an object to an image. The serialization image is opaque
+    object for the client.
 
-  virtual bool materialize(uint serialization_version,
-                           const String *serialization);
+    @param[in]  thd     Thread context.
+    @param[out] image   Serialization image.
 
-  virtual bool do_execute(THD *thd);
+    @return Error status.
+  */
+  virtual bool serialize(THD *thd, String *image);
 
-  const String *describe();
+  /**
+    Create an object persistently in the database.
 
-  const String *build_serialization();
+    @param[in]  thd     Thread context.
 
-  /*
-    The get_db_name primitive is not used for tablespaces.
+    @return Error status.
+  */
+  virtual bool create(THD *thd);
+
+  /**
+    Drop an object in the database.
+
+    @param[in]  thd     Thread context.
+
+    @return Error status.
   */
-  const String *get_db_name() { return 0; }
+  virtual bool drop(THD *thd);
 
-  const String* get_name()
-  { return &m_ts_name; }
+public:
+  /**
+    Read the object state from a given serialization image and restores
+    object state to the point, where it can be created persistently in the
+    database.
 
-  const String* get_datafile()
-  { return &m_datafile; }
+    @param[in] image_version The version of the serialization format.
+    @param[in] image         Buffer contained serialized object.
 
-  const String* get_comments()
-  { return &m_comments; }
+    @return Error status.
+      @retval FALSE on success.
+      @retval TRUE on error.
+  */
+  virtual bool init_from_image(uint image_version, const String *image);
 
-  const String* get_engine()
-  { return &m_engine; }
+protected:
+  /**
+    A primitive implementing @c serialize() method.
+  */
+  virtual bool do_serialize(THD *thd, Out_stream &out_stream) = 0;
 
-  void set_datafile(const String *df)
-  { m_datafile.copy(*df); }
+  /**
+    A primitive implementing @c init_from_image() method.
+  */
+  virtual bool do_init_from_image(In_stream *is);
 
-  void set_comments(const String *c)
-  { m_comments.copy(*c); }
+  virtual void build_drop_statement(String_stream &s_stream) const = 0;
 
-  void set_engine(const String *engine)
-  { m_engine.copy(*engine); }
+protected:
+  MEM_ROOT m_mem_root; /* This mem-root is for keeping stmt list. */
+  List<LEX_STRING> m_stmt_list;
 
-private:
-  // These attributes are to be used only for serialization.
-  String m_ts_name;
-  String m_datafile;
-  String m_comments;
-  String m_engine;
+protected:
+  /* These attributes are to be used only for serialization. */
+  String m_id; //< identify object
 
-  // Drop is not supported by this object.
-  bool drop(THD *thd)
-  { return 0; }
+protected:
+  inline Abstract_obj(LEX_STRING id);
+
+  virtual inline ~Abstract_obj();
 
 private:
-  // These attributes are to be used only for materialization.
-  String m_create_stmt;
+  Abstract_obj(const Abstract_obj &);
+  Abstract_obj &operator =(const Abstract_obj &);
 };
 
-/**
-   @class DbGrantObj
+///////////////////////////////////////////////////////////////////////////
 
-   This class provides an abstraction to database-level grants.
-   This class will permit the recording and replaying of these
-   grants.
-*/
-class DbGrantObj : public Obj
+inline Abstract_obj::Abstract_obj(LEX_STRING id)
 {
-public:
-  DbGrantObj(const String *grantee,
-             const String *db_name,
-             const String *priv_type);
+  init_sql_alloc(&m_mem_root, ALLOC_ROOT_MIN_BLOCK_SIZE, 0);
 
-public:
-  virtual bool materialize(uint serialization_version,
-                           const String *serialization);
+  if (id.str && id.length)
+    m_id.copy(id.str, id.length, system_charset_info);
+  else
+    m_id.length(0);
+}
 
-  const String* get_name()
-  {
-    return &m_name;
-  }
+///////////////////////////////////////////////////////////////////////////
 
-  const String *get_db_name()
-  {
-    return &m_db_name;
-  }
+inline Abstract_obj::~Abstract_obj()
+{
+  free_root(&m_mem_root, MYF(0));
+}
 
-  const String *get_priv_type()
-  {
-    return &m_priv_type;
-  }
+///////////////////////////////////////////////////////////////////////////
 
-protected:
-  // These attributes are to be used only for serialization.
-  String m_db_name;   ///< corresponds with TABLE_SCHEMA in IS tables.
-  String m_name;      ///< name used to list in catalog.
-  String m_grantee;   ///< corresponds with GRANTEE in IS tables.
-  String m_priv_type; ///< corresponds with PRIVILEGE_TYPE in IS tables.
+/**
+  Serialize object state into a buffer. The buffer actually should be a
+  binary buffer. String class is used here just because we don't have
+  convenient primitive for binary buffers.
 
-  bool drop(THD *thd) { return 0; };  // Drop not supported.
-  virtual bool do_execute(THD *thd);
+  Serialization format is opaque to the client, i.e. the client should
+  not make any assumptions about the format or the content of the
+  returned buffer.
 
-private:
-  virtual bool do_serialize(THD *thd, String *serialization);
-  // These attributes are to be used only for materialization.
-  String m_grant_stmt;
-};
+  Serialization format can be changed in the future versions. However,
+  the server must be able to materialize objects coded in any previous
+  formats.
 
-/**
-   @class TblGrantObj
+  @param[in] thd              Server thread context.
+  @param[in] image Buffer to serialize the object
 
-   This class provides an abstraction to table-level and routine-level grants.
-   This class will permit the recording and replaying of these
-   grants.
+  @return Error status.
+    @retval FALSE on success.
+    @retval TRUE on error.
+
+  @note The real work is done inside @c do_serialize() primitive which should be
+  defied in derived classes. This method prepares appropriate context and calls
+  the primitive.
 */
-class TblGrantObj : public DbGrantObj
+
+bool Abstract_obj::serialize(THD *thd, String *image)
 {
-public:
-  TblGrantObj(const String *grantee,
-              const String *db_name,
-              const String *table_name,
-              const String *priv_type);
+  ulong sql_mode_saved= thd->variables.sql_mode;
+  thd->variables.sql_mode= 0;
 
-public:
+  Out_stream out_stream(image);
 
-  const String *get_table_name()
-  {
-    return &m_table_name;
-  }
-
-protected:
-  // These attributes are to be used only for serialization.
-  String m_table_name; ///< corresponds with TABLE_NAME in IS tables.
+  bool ret= do_serialize(thd, out_stream);
 
+  thd->variables.sql_mode= sql_mode_saved;
 
-private:
-  virtual bool do_serialize(THD *thd, String *serialization);
+  return ret;
+}
 
-  // These attributes are to be used only for materialization.
-  String m_grant_stmt;
-};
+///////////////////////////////////////////////////////////////////////////
 
 /**
-   @class ColGrantObj
+  Create the object in the database.
+
+  @param[in] thd              Server thread context.
 
-   This class provides an abstraction to column-level grants.
-   This class will permit the recording and replaying of these
-   grants.
+  @return Error status.
+    @retval FALSE on success.
+    @retval TRUE on error.
 */
-class ColGrantObj : public TblGrantObj
+
+bool Abstract_obj::create(THD *thd)
 {
-public:
-  ColGrantObj(const String *grantee,
-              const String *db_name,
-              const String *table_name,
-              const String *col_name,
-              const String *priv_type);
+  bool rc= FALSE;
+  List_iterator_fast<LEX_STRING> it(m_stmt_list);
+  LEX_STRING *sql_text;
+  Si_session_context session_context;
 
-public:
+  DBUG_ENTER("Abstract_obj::create");
+
+  /*
+    Drop the object if it exists first of all.
+
+    @note this semantics will be changed in the future.
+    That's why drop() is a public separate operation of Obj.
+  */
+  drop(thd);
 
-  const String *get_col_name()
+  /*
+    Now, proceed with creating the object.
+  */
+  session_context.save_si_ctx(thd);
+  session_context.reset_si_ctx(thd);
+
+  /* Allow to execute DDL operations. */
+  ::obs::ddl_blocker_exception_on(thd);
+
+  /* Run queries from the serialization image. */
+  while ((sql_text= it++))
   {
-    return &m_col_name;
+    Ed_result ed_result;
+
+    rc= mysql_execute_direct(thd, *sql_text, &ed_result);
+
+    /* Push warnings on the THD error stack. */
+    copy_warnings(thd, &ed_result.get_warnings());
+
+    if (rc)
+      break;
   }
 
-protected:
-  // These attributes are to be used only for serialization.
-  String m_col_name; ///< corresponds with COLUMN_NAME in IS tables.
+  /* Disable further DDL execution. */
+  ::obs::ddl_blocker_exception_off(thd);
 
-private:
-  virtual bool do_serialize(THD *thd, String *serialization);
+  session_context.restore_si_ctx(thd);
 
-  // These attributes are to be used only for materialization.
-  String m_grant_stmt;
-};
+  DBUG_RETURN(rc);
+}
 
 ///////////////////////////////////////////////////////////////////////////
 
-//
-// Implementation: iterator impl classes.
-//
+/**
+  Drop the object in the database.
 
-///////////////////////////////////////////////////////////////////////////
+  @param[in] thd              Server thread context.
+
+  @return Error status.
+    @retval FALSE on success.
+    @retval TRUE on error.
+*/
 
-class InformationSchemaIterator : public Obj_iterator
+bool Abstract_obj::drop(THD *thd)
 {
-public:
-  static bool prepare_is_table(
-    THD *thd,
-    TABLE **is_table,
-    handler **ha,
-    my_bitmap_map **orig_columns,
-    enum_schema_tables is_table_idx,
-    List<LEX_STRING> db_list);
-
-public:
-  InformationSchemaIterator(THD *thd,
-                            TABLE *is_table,
-                            handler *ha,
-                            my_bitmap_map *orig_columns)
-    :
-      m_thd(thd),
-      m_is_table(is_table),
-      m_ha(ha),
-      m_orig_columns(orig_columns)
-  { }
+  String_stream s_stream;
+  const LEX_STRING *sql_text;
 
-  virtual ~InformationSchemaIterator();
+  DBUG_ENTER("Abstract_obj::drop");
 
-public:
-  virtual Obj *next();
+  build_drop_statement(s_stream);
+  sql_text= s_stream.lex_string();
 
-protected:
-  virtual Obj *create_obj(TABLE *t) = 0;
-  THD *m_thd;
+  if (!sql_text->str || !sql_text->length)
+    DBUG_RETURN(FALSE);
 
-private:
-  TABLE *m_is_table;
-  handler *m_ha;
-  my_bitmap_map *m_orig_columns;
+  Si_session_context session_context;
 
-};
+  session_context.save_si_ctx(thd);
+  session_context.reset_si_ctx(thd);
 
-///////////////////////////////////////////////////////////////////////////
+  /* Allow to execute DDL operations. */
+  ::obs::ddl_blocker_exception_on(thd);
 
-class ObjIteratorDummyImpl : Obj_iterator
-{
-public:
-  ObjIteratorDummyImpl() { return; }
-  virtual ~ObjIteratorDummyImpl() { return; }
-  virtual Obj *next() { return NULL; }
+  Ed_result ed_result;
 
-protected:
-  virtual Obj *create_obj(TABLE *t) { return NULL; }
+  /* Execute DDL operation. */
+  bool rc= mysql_execute_direct(thd, *sql_text, &ed_result);
 
-};
+  /* Disable further DDL execution. */
+  ::obs::ddl_blocker_exception_off(thd);
+
+  session_context.restore_si_ctx(thd);
+
+  DBUG_RETURN(rc);
+}
 
 ///////////////////////////////////////////////////////////////////////////
-class DatabaseIterator : public InformationSchemaIterator
+
+bool Abstract_obj::init_from_image(uint image_version, const String *image)
 {
-public:
-  DatabaseIterator(THD *thd,
-                   TABLE *is_table,
-                   handler *ha,
-                   my_bitmap_map *orig_columns) :
-    InformationSchemaIterator(thd, is_table, ha, orig_columns)
-  { }
+  In_stream is(image_version, image);
 
-protected:
-  virtual DatabaseObj *create_obj(TABLE *t);
-};
+  return do_init_from_image(&is);
+}
 
 ///////////////////////////////////////////////////////////////////////////
 
-class DbTablesIterator : public InformationSchemaIterator
+bool Abstract_obj::do_init_from_image(In_stream *is)
 {
-public:
-  DbTablesIterator(THD *thd,
-                   const String *db_name,
-                   TABLE *is_table,
-                   handler *ha,
-                   my_bitmap_map *orig_columns) :
-    InformationSchemaIterator(thd, is_table, ha, orig_columns)
+  LEX_STRING sql_text;
+  while (! is->next(&sql_text))
   {
-    m_db_name.copy(*db_name);
-  }
+    LEX_STRING *sql_text_root= (LEX_STRING *) alloc_root(&m_mem_root,
+                                                         sizeof (LEX_STRING));
 
-protected:
-  virtual TableObj *create_obj(TABLE *t);
-
-  virtual bool is_type_accepted(const String *type) const;
+    if (!sql_text_root)
+      return TRUE;
 
-  virtual bool is_engine_accepted(const String *engine) const;
+    sql_text_root->str= strmake_root(&m_mem_root,
+                                     sql_text.str, sql_text.length);
+    sql_text_root->length= sql_text.length;
 
-  virtual TableObj *create_table_obj(const String *db_name,
-                                     const String *table_name) const;
+    if (!sql_text_root->str || m_stmt_list.push_back(sql_text_root))
+      return TRUE;
+  }
 
-private:
-  String m_db_name;
-};
+  return FALSE;
+}
 
 ///////////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////
 
-class DbViewsIterator : public DbTablesIterator
+/**
+  @class Database_obj
+
+  This class provides an abstraction to a database object for creation and
+  capture of the creation data.
+*/
+
+class Database_obj : public Abstract_obj
 {
 public:
-  DbViewsIterator(THD *thd,
-                  const String *db_name,
-                  TABLE *is_tables,
-                  handler *ha,
-                  my_bitmap_map *orig_columns)
-    : DbTablesIterator(thd, db_name, is_tables, ha, orig_columns)
+  inline Database_obj(const Ed_row &ed_row)
+    : Abstract_obj(ed_row[0] /* database name */)
   { }
 
-protected:
-  virtual bool is_type_accepted(const String *type) const;
+  inline Database_obj(LEX_STRING db_name)
+    : Abstract_obj(db_name)
+  { }
 
-  virtual bool is_engine_accepted(const String *engine) const
-  {
-    return true;
-  }
+public:
+  virtual inline const String *get_db_name() const { return get_name(); }
 
-  virtual TableObj *create_table_obj(const String *db_name,
-                                     const String *table_name) const;
+private:
+  virtual bool do_serialize(THD *thd, Out_stream &out_stream);
+  virtual void build_drop_statement(String_stream &s_stream) const;
 };
 
 ///////////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////
+
+/**
+  @class Database_item_obj
 
-class DbTriggerIterator : public InformationSchemaIterator
+  This class is a base class for all classes representing objects, which
+  belong to a Database.
+*/
+
+class Database_item_obj : public Abstract_obj
 {
 public:
-  DbTriggerIterator(THD *thd,
-                    const String *db_name,
-                    TABLE *is_table,
-                    handler *ha,
-                    my_bitmap_map *orig_columns) :
-    InformationSchemaIterator(thd, is_table, ha, orig_columns)
+  inline Database_item_obj(LEX_STRING db_name, LEX_STRING object_name)
+    : Abstract_obj(object_name)
   {
-    m_db_name.copy(*db_name);
+    if (db_name.str && db_name.length)
+      m_db_name.copy(db_name.str, db_name.length, system_charset_info);
+    else
+      m_db_name.length(0);
   }
 
+public:
+  virtual inline const String *get_db_name() const { return &m_db_name; }
+
 protected:
-  virtual TriggerObj *create_obj(TABLE *t);
+  virtual const LEX_STRING *get_type_name() const = 0;
+  virtual void build_drop_statement(String_stream &s_stream) const;
 
-private:
+protected:
   String m_db_name;
 };
 
 ///////////////////////////////////////////////////////////////////////////
 
-class DbStoredProcIterator : public InformationSchemaIterator
+void Database_item_obj::build_drop_statement(String_stream &s_stream) const
 {
-public:
-  DbStoredProcIterator(THD *thd,
-                       const String *db_name,
-                       TABLE *is_table,
-                       handler *ha,
-                       my_bitmap_map *orig_columns) :
-    InformationSchemaIterator(thd, is_table, ha, orig_columns)
-  {
-    m_db_name.copy(*db_name);
-  }
+  s_stream <<
+    "DROP " << get_type_name() << " IF EXISTS `" << get_name() << "`";
+}
 
-protected:
-  virtual Obj *create_obj(TABLE *t);
+///////////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////
 
-  virtual bool check_type(const String *sr_type) const;
+/**
+  @class Table_obj
 
-  virtual Obj *create_sr_object(const String *db_name,
-                                const String *sr_name);
+  This class provides an abstraction to a table object for creation and
+  capture of the creation data.
+*/
 
+class Table_obj : public Database_item_obj
+{
 private:
-  String m_db_name;
-};
+  static const LEX_STRING TYPE_NAME;
 
-class DbGrantIterator : public InformationSchemaIterator
-{
 public:
-  DbGrantIterator(THD *thd,
-                 const String *db_name,
-                 TABLE *is_table,
-                 handler *ha,
-                 my_bitmap_map *orig_columns) :
-    InformationSchemaIterator(thd, is_table, ha, orig_columns)
-  {
-    m_db_name.copy(*db_name);
-  }
+  inline Table_obj(const Ed_row &ed_row)
+    : Database_item_obj(ed_row[0], /* database name */
+                        ed_row[1]) /* table name */
+  { }
 
-protected:
-  virtual DbGrantObj *create_obj(TABLE *t);
+  inline Table_obj(LEX_STRING db_name, LEX_STRING table_name)
+    : Database_item_obj(db_name, table_name)
+  { }
 
 private:
-  String m_db_name;
+  virtual bool do_serialize(THD *thd, Out_stream &out_stream);
+
+  virtual inline const LEX_STRING *get_type_name() const
+  { return &Table_obj::TYPE_NAME; }
 };
 
-class TblGrantIterator : public InformationSchemaIterator
-{
-public:
-  TblGrantIterator(THD *thd,
-                  const String *db_name,
-                  TABLE *is_table,
-                  handler *ha,
-                  my_bitmap_map *orig_columns) :
-    InformationSchemaIterator(thd, is_table, ha, orig_columns)
-  {
-    m_db_name.copy(*db_name);
-  }
+///////////////////////////////////////////////////////////////////////////
 
-protected:
-  virtual TblGrantObj *create_obj(TABLE *t);
+const LEX_STRING Table_obj::TYPE_NAME= LXS_INIT("TABLE");
 
-private:
-  String m_db_name;
-};
+///////////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////
 
-class ColGrantIterator : public InformationSchemaIterator
+/**
+  @class View_obj
+
+  This class provides an abstraction to a view object for creation and
+  capture of the creation data.
+*/
+
+class View_obj : public Database_item_obj
 {
+private:
+  static const LEX_STRING TYPE_NAME;
+
 public:
-  ColGrantIterator(THD *thd,
-                  const String *db_name,
-                  TABLE *is_table,
-                  handler *ha,
-                  my_bitmap_map *orig_columns) :
-    InformationSchemaIterator(thd, is_table, ha, orig_columns)
-  {
-    m_db_name.copy(*db_name);
-  }
+  inline View_obj(const Ed_row &ed_row)
+    : Database_item_obj(ed_row[0], /* schema name */
+                        ed_row[1]) /* view name */
+  { }
 
-protected:
-  virtual ColGrantObj *create_obj(TABLE *t);
+  inline View_obj(LEX_STRING db_name, LEX_STRING view_name)
+    : Database_item_obj(db_name, view_name)
+  { }
 
 private:
-  String m_db_name;
+  virtual bool do_serialize(THD *thd, Out_stream &out_stream);
+
+  virtual inline const LEX_STRING *get_type_name() const
+  { return &View_obj::TYPE_NAME; }
 };
 
+///////////////////////////////////////////////////////////////////////////
+
+const LEX_STRING View_obj::TYPE_NAME= LXS_INIT("VIEW");
 
 ///////////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////
 
-class DbStoredFuncIterator : public DbStoredProcIterator
+/**
+  @class Stored_program_obj
+
+  This is a base class for stored program objects: stored procedures,
+  stored functions, triggers, events.
+*/
+
+class Stored_program_obj : public Database_item_obj
 {
 public:
-  DbStoredFuncIterator(THD *thd,
-                       const String *db_name,
-                       TABLE *is_table,
-                       handler *ha,
-                       my_bitmap_map *orig_columns) :
-    DbStoredProcIterator(thd, db_name, is_table, ha, orig_columns)
+  inline Stored_program_obj(LEX_STRING db_name, LEX_STRING sp_name)
+    : Database_item_obj(db_name, sp_name)
   { }
 
 protected:
-  virtual bool check_type(const String *sr_type) const;
+  virtual const LEX_STRING *get_create_stmt(Ed_row *row) = 0;
+  virtual void dump_header(Ed_row *row, Out_stream &out_stream) = 0;
 
-  virtual Obj *create_sr_object(const String *db_name,
-                                const String *sr_name);
+private:
+  virtual bool do_serialize(THD *thd, Out_stream &out_stream);
 };
 
 ///////////////////////////////////////////////////////////////////////////
-#ifdef HAVE_EVENT_SCHEDULER
-class DbEventIterator : public InformationSchemaIterator
+///////////////////////////////////////////////////////////////////////////
+
+/**
+  @class Stored_routine_obj
+
+  This is a base class for stored routine objects: stored procedures,
+  stored functions, triggers.
+*/
+
+class Stored_routine_obj : public Stored_program_obj
 {
 public:
-  DbEventIterator(THD *thd,
-                  const String *db_name,
-                  TABLE *is_table,
-                  handler *ha,
-                  my_bitmap_map *orig_columns) :
-    InformationSchemaIterator(thd, is_table, ha, orig_columns)
-  {
-    m_db_name.copy(*db_name);
-  }
+  inline Stored_routine_obj(LEX_STRING db_name, LEX_STRING sr_name)
+    : Stored_program_obj(db_name, sr_name)
+  { }
 
 protected:
-  virtual EventObj *create_obj(TABLE *t);
-
-private:
-  String m_db_name;
+  virtual const LEX_STRING *get_create_stmt(Ed_row *row);
+  virtual void dump_header(Ed_row *row, Out_stream &out_stream);
 };
-#endif
 
 ///////////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////
 
-class ViewBaseObjectsIterator : public Obj_iterator
-{
-public:
-  enum IteratorType
-  {
-    GET_BASE_TABLES,
-    GET_BASE_VIEWS
-  };
-
-public:
-  virtual ~ViewBaseObjectsIterator();
+/**
+  @class Trigger_obj
 
-public:
-  virtual TableObj *next();
+  This class provides an abstraction to a trigger object for creation and
+  capture of the creation data.
+*/
 
+class Trigger_obj : public Stored_routine_obj
+{
 private:
-  static ViewBaseObjectsIterator *create(THD *thd,
-                                         const String *db_name,
-                                         const String *view_name,
-                                         IteratorType iterator_type );
+  static const LEX_STRING TYPE_NAME;
 
-private:
-  ViewBaseObjectsIterator(HASH *table_names);
+public:
+  inline Trigger_obj(const Ed_row &ed_row)
+    : Stored_routine_obj(ed_row[0], ed_row[1])
+  { }
 
-private:
-  HASH *m_table_names;
-  uint m_cur_idx;
+  inline Trigger_obj(LEX_STRING db_name, LEX_STRING trigger_name)
+    : Stored_routine_obj(db_name, trigger_name)
+  { }
 
 private:
-  friend Obj_iterator *get_view_base_tables(THD *,
-                                            const String *,
-                                            const String *);
-
-  friend Obj_iterator *get_view_base_views(THD *,
-                                           const String *,
-                                           const String *);
+  virtual inline const LEX_STRING *get_type_name() const
+  { return &Trigger_obj::TYPE_NAME; }
 };
 
 ///////////////////////////////////////////////////////////////////////////
 
-//
-// Implementation: InformationSchemaIterator class.
-//
+const LEX_STRING Trigger_obj::TYPE_NAME= LXS_INIT("TRIGGER");
 
 ///////////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////
+
+/**
+  @class Stored_proc_obj
 
-bool InformationSchemaIterator::prepare_is_table(
-  THD *thd,
-  TABLE **is_table,
-  handler **ha,
-  my_bitmap_map **orig_columns,
-  enum_schema_tables is_table_idx,
-  List<LEX_STRING> db_list)
+  This class provides an abstraction to a stored procedure object for creation
+  and capture of the creation data.
+*/
+
+class Stored_proc_obj : public Stored_routine_obj
 {
-  ST_SCHEMA_TABLE *st= get_schema_table(is_table_idx);
-  if (!(*is_table= open_schema_table(thd, st, &db_list)))
-    return TRUE;
+private:
+  static const LEX_STRING TYPE_NAME;
 
-  *ha= (*is_table)->file;
+public:
+  inline Stored_proc_obj(const Ed_row &ed_row)
+    : Stored_routine_obj(ed_row[0], ed_row[1])
+  { }
 
-  if (!*ha)
-  {
-    free_tmp_table(thd, *is_table);
-    return TRUE;
-  }
+  inline Stored_proc_obj(LEX_STRING db_name, LEX_STRING sp_name)
+    : Stored_routine_obj(db_name, sp_name)
+  { }
 
-  *orig_columns=
-    dbug_tmp_use_all_columns(*is_table, (*is_table)->read_set);
+private:
+  virtual inline const LEX_STRING *get_type_name() const
+  { return &Stored_proc_obj::TYPE_NAME; }
+};
 
-  if ((*ha)->ha_rnd_init(TRUE))
-  {
-    dbug_tmp_restore_column_map((*is_table)->read_set, *orig_columns);
-    free_tmp_table(thd, *is_table);
-    return TRUE;
-  }
+///////////////////////////////////////////////////////////////////////////
 
-  return FALSE;
-}
+const LEX_STRING Stored_proc_obj::TYPE_NAME= LXS_INIT("PROCEDURE");
 
-InformationSchemaIterator::~InformationSchemaIterator()
-{
-  m_ha->ha_rnd_end();
+///////////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////
 
-  dbug_tmp_restore_column_map(m_is_table->read_set, m_orig_columns);
-  free_tmp_table(m_thd, m_is_table);
-}
+/**
+  @class Stored_func_obj
 
-Obj *InformationSchemaIterator::next()
+  This class provides an abstraction to a stored function object for creation
+  and capture of the creation data.
+*/
+
+class Stored_func_obj : public Stored_routine_obj
 {
-  while (true)
-  {
-    if (m_ha->rnd_next(m_is_table->record[0]))
-      return NULL;
+private:
+  static const LEX_STRING TYPE_NAME;
 
-    Obj *obj= create_obj(m_is_table);
+public:
+  inline Stored_func_obj(const Ed_row &ed_row)
+    : Stored_routine_obj(ed_row[0], ed_row[1])
+  { }
 
-    if (obj)
-      return obj;
-  }
-}
+  inline Stored_func_obj(LEX_STRING db_name, LEX_STRING sf_name)
+    : Stored_routine_obj(db_name, sf_name)
+  { }
+
+private:
+  virtual inline const LEX_STRING *get_type_name() const
+  { return &Stored_func_obj::TYPE_NAME; }
+};
 
 ///////////////////////////////////////////////////////////////////////////
 
-//
-// Implementation: DatabaseIterator class.
-//
+const LEX_STRING Stored_func_obj::TYPE_NAME= LXS_INIT("FUNCTION");
 
 ///////////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////
 
-DatabaseObj* DatabaseIterator::create_obj(TABLE *t)
+#ifdef HAVE_EVENT_SCHEDULER
+
+/**
+  @class Event_obj
+
+  This class provides an abstraction to a event object for creation and capture
+  of the creation data.
+*/
+
+class Event_obj : public Stored_program_obj
 {
-  String name;
+private:
+  static const LEX_STRING TYPE_NAME;
+
+public:
+  inline Event_obj(const Ed_row &ed_row)
+    : Stored_program_obj(ed_row[0], ed_row[1])
+  { }
+
+  inline Event_obj(LEX_STRING db_name, LEX_STRING event_name)
+    : Stored_program_obj(db_name, event_name)
+  { }
 
-  t->field[1]->val_str(&name);
+private:
+  virtual inline const LEX_STRING *get_type_name() const
+  { return &Event_obj::TYPE_NAME; }
 
-  DBUG_PRINT("DatabaseIterator::next", (" Found database %s", name.ptr()));
+  virtual inline const LEX_STRING *get_create_stmt(Ed_row *row)
+  { return row->get_column(3); }
 
-  return new DatabaseObj(&name);
-}
+  virtual void dump_header(Ed_row *row, Out_stream &out_stream);
+};
 
 ///////////////////////////////////////////////////////////////////////////
 
-//
-// Implementation: DbTablesIterator class.
-//
+const LEX_STRING Event_obj::TYPE_NAME= LXS_INIT("EVENT");
+
+#endif // HAVE_EVENT_SCHEDULER
 
 ///////////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////
 
-TableObj* DbTablesIterator::create_obj(TABLE *t)
-{
-  String table_name;
-  String db_name;
-  String type;
-  String engine;
+/**
+  @class Tablespace_obj
 
-  t->field[1]->val_str(&db_name);
-  t->field[2]->val_str(&table_name);
-  t->field[3]->val_str(&type);
-  t->field[4]->val_str(&engine);
+  This class provides an abstraction to a user object for creation and
+  capture of the creation data.
+*/
 
-  // Skip tables not from the given database.
+class Tablespace_obj : public Abstract_obj
+{
+public:
+  Tablespace_obj(LEX_STRING ts_name,
+                 LEX_STRING comment,
+                 LEX_STRING data_file_name,
+                 LEX_STRING engine_name);
 
-  if (db_name != m_db_name)
-    return NULL;
+  Tablespace_obj(LEX_STRING ts_name);
 
-  // Skip tables/views depending on enumerate_views flag.
+public:
+  const String *get_description();
 
-  if (!is_type_accepted(&type))
-    return NULL;
+public:
+  virtual bool init_from_image(uint image_version, const String *image);
 
-  // TODO: actually, Backup Kernel needs to know also tables with
-  // invalid/empty engines. It is required so that Backup Kernel can throw
-  // a warning to the user.
+private:
+  virtual bool do_serialize(THD *thd, Out_stream &out_stream);
+  virtual void build_drop_statement(String_stream &s_stream) const;
 
-  if (!is_engine_accepted(&engine))
-    return NULL;
+private:
+  /* These attributes are to be used only for serialization. */
+  String m_comment;
+  String m_data_file_name;
+  String m_engine;
 
-  DBUG_PRINT("DbTablesIterator::next", (" Found table %s.%s",
-                                        db_name.ptr(), table_name.ptr()));
+private:
+  String m_description;
+};
 
-  return create_table_obj(&db_name, &table_name);
-}
+///////////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////
 
-bool DbTablesIterator::is_type_accepted(const String *type) const
-{
-  return my_strcasecmp(system_charset_info,
-                       ((String *) type)->c_ptr_safe(), "BASE TABLE") == 0;
-}
+/**
+  @class Grant_obj
 
-bool DbTablesIterator::is_engine_accepted(const String *engine) const
-{
-  return engine->length() > 0;
-}
+  This class provides an abstraction to grants. This class will permit the
+  recording and replaying of these grants.
+*/
 
-TableObj *DbTablesIterator::create_table_obj(const String *db_name,
-                                             const String *table_name) const
+class Grant_obj : public Abstract_obj
 {
-  return new TableObj(db_name, table_name, false);
-}
+public:
+  static void generate_unique_grant_id(const String *user_name, String *id);
 
-///////////////////////////////////////////////////////////////////////////
+public:
+  Grant_obj(LEX_STRING grant_id);
+  Grant_obj(const Ed_row &ed_row);
 
-//
-// Implementation: DbViewsIterator class.
-//
+public:
+  inline const String *get_user_name() const { return &m_user_name; }
+  inline const String *get_grant_info() const { return &m_grant_info; }
 
-///////////////////////////////////////////////////////////////////////////
+private:
+  virtual bool do_init_from_image(In_stream *is);
+  inline virtual void build_drop_statement(String_stream &s_stream) const
+  { }
 
-bool DbViewsIterator::is_type_accepted(const String *type) const
-{
-  return my_strcasecmp(system_charset_info,
-                       ((String *) type)->c_ptr_safe(), "VIEW") == 0;
-}
+private:
+  /* These attributes are to be used only for serialization. */
+  String m_user_name;
+  String m_grant_info; //< contains privilege definition
 
-TableObj *DbViewsIterator::create_table_obj(const String *db_name,
-                                            const String *table_name) const
-{
-  return new TableObj(db_name, table_name, true);
-}
+private:
+  virtual bool do_serialize(THD *thd, Out_stream &out_stream);
+};
 
 ///////////////////////////////////////////////////////////////////////////
-
-//
-// Implementation: DbTriggerIterator class.
-//
-
 ///////////////////////////////////////////////////////////////////////////
 
-TriggerObj *DbTriggerIterator::create_obj(TABLE *t)
+template <typename Iterator>
+Iterator *create_row_set_iterator(THD *thd, const LEX_STRING *query)
 {
-  String db_name;
-  String trigger_name;
+  Ed_result *ed_result= new Ed_result();
 
-  t->field[1]->val_str(&db_name);
-  t->field[2]->val_str(&trigger_name);
-
-  // Skip triggers not from the given database.
-
-  if (db_name != m_db_name)
+  if (run_service_interface_sql(thd, query, ed_result) ||
+      ed_result->get_warnings().elements > 0)
+  {
+    /* There should be no warnings. */
+    delete ed_result;
     return NULL;
+  }
+
+  /* The result must contain only one result-set. */
+  DBUG_ASSERT(ed_result->elements == 1);
 
-  return new TriggerObj(&db_name, &trigger_name);
+  return new Iterator(ed_result);
 }
 
 ///////////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////
 
-//
-// Implementation: DbStoredProcIterator class.
-//
+/**
+  @class Ed_result_set_iterator
 
-///////////////////////////////////////////////////////////////////////////
+  This is an implementation of Obj_iterator, which creates objects from a
+  result set (represented by an instance of Ed_result).
+*/
 
-Obj *DbStoredProcIterator::create_obj(TABLE *t)
+template <typename Obj_type>
+class Ed_result_set_iterator : public Obj_iterator
 {
-  String db_name;
-  String sr_name;
-  String sr_type;
+public:
+  inline Ed_result_set_iterator(Ed_result *ed_result);
+  inline ~Ed_result_set_iterator();
 
-  t->field[2]->val_str(&db_name);
-  t->field[3]->val_str(&sr_name);
-  t->field[4]->val_str(&sr_type);
+public:
+  virtual Obj *next();
 
-  // Skip stored procedure not from the given database.
+private:
+  Ed_result *m_ed_result;
+  List_iterator_fast<Ed_row> m_row_it;
+};
 
-  if (db_name != m_db_name)
-    return NULL;
+///////////////////////////////////////////////////////////////////////////
 
-  if (!check_type(&sr_type))
-    return NULL;
+template <typename Obj_type>
+inline
+Ed_result_set_iterator<Obj_type>::
+Ed_result_set_iterator(Ed_result *ed_result)
+  : m_ed_result(ed_result),
+    m_row_it(*ed_result->get_cur_result_set()->data())
+{ }
 
-  return create_sr_object(&db_name, &sr_name);
-}
+///////////////////////////////////////////////////////////////////////////
 
-bool DbStoredProcIterator::check_type(const String *sr_type) const
+template <typename Obj_type>
+inline
+Ed_result_set_iterator<Obj_type>::~Ed_result_set_iterator()
 {
-  return
-    my_strcasecmp(system_charset_info,
-                  ((String *) sr_type)->c_ptr_safe(),
-                  "PROCEDURE") == 0;
+  delete m_ed_result;
 }
 
-Obj *DbStoredProcIterator::create_sr_object(const String *db_name,
-                                            const String *sr_name)
+
+template <typename Obj_type>
+Obj *
+Ed_result_set_iterator<Obj_type>::next()
 {
-  return new StoredProcObj(db_name, sr_name);
-}
+  Ed_row *ed_row= m_row_it++;
 
-///////////////////////////////////////////////////////////////////////////
+  if (!ed_row)
+    return NULL;
 
-//
-// Implementation: DbStoredFuncIterator class.
-//
+  return new Obj_type(*ed_row);
+}
 
 ///////////////////////////////////////////////////////////////////////////
 
-bool DbStoredFuncIterator::check_type(const String *sr_type) const
-{
-  return
-    my_strcasecmp(system_charset_info,
-                  ((String *) sr_type)->c_ptr_safe(),
-                  "FUNCTION") == 0;
-}
+typedef Ed_result_set_iterator<Database_obj>    Database_iterator;
+typedef Ed_result_set_iterator<Table_obj>       Db_tables_iterator;
+typedef Ed_result_set_iterator<View_obj>        Db_views_iterator;
+typedef Ed_result_set_iterator<Trigger_obj>     Db_trigger_iterator;
+typedef Ed_result_set_iterator<Stored_proc_obj> Db_stored_proc_iterator;
+typedef Ed_result_set_iterator<Stored_func_obj> Db_stored_func_iterator;
+#ifdef HAVE_EVENT_SCHEDULER
+typedef Ed_result_set_iterator<Event_obj>       Db_event_iterator;
+#endif
+typedef Ed_result_set_iterator<Grant_obj>       Grant_iterator;
 
-Obj *DbStoredFuncIterator::create_sr_object(const String *db_name,
-                                            const String *sr_name)
-{
-  return new StoredFuncObj(db_name, sr_name);
-}
+///////////////////////////////////////////////////////////////////////////
 
+#ifdef HAVE_EXPLICIT_TEMPLATE_INSTANTIATION
+template class Ed_result_set_iterator<Database_obj>;
+template class Ed_result_set_iterator<Table_obj>;
+template class Ed_result_set_iterator<View_obj>;
+template class Ed_result_set_iterator<Trigger_obj>;
+template class Ed_result_set_iterator<Stored_proc_obj>;
+template class Ed_result_set_iterator<Stored_func_obj>;
 #ifdef HAVE_EVENT_SCHEDULER
-///////////////////////////////////////////////////////////////////////////
+template class Ed_result_set_iterator<Event_obj>;
+#endif
+template class Ed_result_set_iterator<Grant_obj>;
 
-//
-// Implementation: DbEventIterator class.
-//
+template
+Database_iterator *
+create_row_set_iterator<Database_iterator>(THD *thd, const LEX_STRING *query);
 
-///////////////////////////////////////////////////////////////////////////
-EventObj *DbEventIterator::create_obj(TABLE *t)
-{
-  String db_name;
-  String event_name;
+template
+Db_tables_iterator *
+create_row_set_iterator<Db_tables_iterator>(THD *thd, const LEX_STRING *query);
 
-  t->field[1]->val_str(&db_name);
-  t->field[2]->val_str(&event_name);
+template
+Db_views_iterator *
+create_row_set_iterator<Db_views_iterator>(THD *thd, const LEX_STRING *query);
 
-  // Skip event not from the given database.
+template
+Db_trigger_iterator *
+create_row_set_iterator<Db_trigger_iterator>(THD *thd, const LEX_STRING *query);
 
-  if (db_name != m_db_name)
-    return NULL;
+template
+Db_stored_proc_iterator *
+create_row_set_iterator<Db_stored_proc_iterator>(THD *thd,
+                                                 const LEX_STRING *query);
 
-  return new EventObj(&db_name, &event_name);
-}
+template
+Db_stored_func_iterator *
+create_row_set_iterator<Db_stored_func_iterator>(THD *thd,
+                                                 const LEX_STRING *query);
+
+#ifdef HAVE_EVENT_SCHEDULER
+template
+Db_event_iterator *
+create_row_set_iterator<Db_event_iterator>(THD *thd, const LEX_STRING *query);
 #endif
 
-///////////////////////////////////////////////////////////////////////////
+template
+Grant_iterator *
+create_row_set_iterator<Grant_iterator>(THD *thd, const LEX_STRING *query);
 
-//
-// Implementation: DbGrantIterator class.
-//
+#endif // HAVE_EXPLICIT_TEMPLATE_INSTANTIATION
 
 ///////////////////////////////////////////////////////////////////////////
 
-DbGrantObj* DbGrantIterator::create_obj(TABLE *t)
+/**
+  @class View_base_iterator
+
+  This is a base implementation of Obj_iterator for the
+  view-dependency-object iterators.
+*/
+
+class View_base_obj_iterator : public Obj_iterator
 {
-  String grantee;   // corresponds with GRANTEE
-  String db_name;   // corresponds with TABLE_SCHEMA
-  String priv_type; // corresponds with PRIVILEGE_TYPE
-
-  t->field[0]->val_str(&grantee);
-  t->field[2]->val_str(&db_name);
-  t->field[3]->val_str(&priv_type);
+public:
+  inline View_base_obj_iterator();
+  virtual inline ~View_base_obj_iterator();
 
-  /*
-    The fill method for SCHEMA_PRIVILEGES does not use the COND portion
-    of the generic fill() method. Thus, we have to do the restriction here.
+public:
+  virtual Obj *next();
 
-    Ensure the only rows sent back from iterator are the ones that match the
-    database specified.
-  */
-  if (db_name == m_db_name)
-  {
-    DBUG_PRINT("DbGrantIterator::create", (" Found grant %s %s %s",
-     db_name.ptr(), grantee.ptr(), priv_type.ptr()));
+  enum enum_base_obj_kind { BASE_TABLE= 0, VIEW };
 
-    /*
-      Include grants for only users that exist at time of backup.
-    */
-    if (check_user_existence(m_thd, &grantee))
-      return new DbGrantObj(&grantee, &db_name, &priv_type);
-    else
-      return NULL;
-  }
-  else
-    return NULL;
-}
+  virtual enum_base_obj_kind get_base_obj_kind() const= 0;
 
-///////////////////////////////////////////////////////////////////////////
+protected:
+  template <typename Iterator>
+  static Iterator *create(THD *thd,
+                          const String *db_name, const String *view_name);
 
-//
-// Implementation: TblGrantIterator class.
-//
+protected:
+  bool init(THD *thd, const String *db_name, const String *view_name);
+
+  virtual Obj *create_obj(const LEX_STRING *db_name,
+                          const LEX_STRING *obj_name)= 0;
+
+private:
+  HASH m_table_names;
+  uint m_cur_idx;
+};
 
 ///////////////////////////////////////////////////////////////////////////
 
-TblGrantObj* TblGrantIterator::create_obj(TABLE *t)
+template <typename Iterator>
+Iterator *View_base_obj_iterator::create(THD *thd,
+                                         const String *db_name,
+                                         const String *view_name)
 {
-  String grantee;   // corresponds with GRANTEE
-  String db_name;   // corresponds with TABLE_SCHEMA
-  String tbl_name;  // corresponds with TABLE_NAME
-  String priv_type; // corresponds with PRIVILEGE_TYPE
-
-  t->field[0]->val_str(&grantee);
-  t->field[2]->val_str(&db_name);
-  t->field[3]->val_str(&tbl_name);
-  t->field[4]->val_str(&priv_type);
-
-  /*
-    The fill method for TABLE_PRIVILEGES does not use the COND portion
-    of the generic fill() method. Thus, we have to do the restriction here.
+  Iterator *it= new Iterator();
 
-    Ensure the only rows sent back from iterator are the ones that match the
-    database specified.
-  */
-  if (db_name == m_db_name)
+  if (it->init(thd, db_name, view_name))
   {
-    DBUG_PRINT("TblGrantIterator::create", (" Found grant %s %s %s %s",
-     db_name.ptr(), grantee.ptr(), tbl_name.ptr(), priv_type.ptr()));
-
-    /*
-      Include grants for only users that exist at time of backup.
-    */
-    if (check_user_existence(m_thd, &grantee))
-      return new TblGrantObj(&grantee, &db_name, &tbl_name, &priv_type);
-    else
-      return NULL;
-  }
-  else
+    delete it;
     return NULL;
+  }
+
+  return it;
 }
 
 ///////////////////////////////////////////////////////////////////////////
 
-//
-// Implementation: ColGrantIterator class.
-//
+inline View_base_obj_iterator::View_base_obj_iterator()
+  :m_cur_idx(0)
+{
+  hash_init(&m_table_names, system_charset_info, 16, 0, 0,
+            get_table_name_key, free_table_name_key,
+            MYF(0));
+}
 
 ///////////////////////////////////////////////////////////////////////////
 
-ColGrantObj* ColGrantIterator::create_obj(TABLE *t)
+inline View_base_obj_iterator::~View_base_obj_iterator()
 {
-  String grantee;   // corresponds with GRANTEE
-  String db_name;   // corresponds with TABLE_SCHEMA
-  String tbl_name;  // corresponds with TABLE_NAME
-  String col_name;  // corresponds with COLUMN_NAME
-  String priv_type; // corresponds with PRIVILEGE_TYPE
-
-  t->field[0]->val_str(&grantee);
-  t->field[2]->val_str(&db_name);
-  t->field[3]->val_str(&tbl_name);
-  t->field[4]->val_str(&col_name);
-  t->field[5]->val_str(&priv_type);
+  hash_free(&m_table_names);
+}
 
-  /*
-    The fill method for COLUMN_PRIVILEGES does not use the COND portion
-    of the generic fill() method. Thus, we have to do the restriction here.
 
-    Ensure the only rows sent back from iterator are the ones that match the
-    database specified.
-  */
-  if (db_name == m_db_name)
-  {
-    DBUG_PRINT("ColGrantIterator::create", (" Found grant %s %s %s %s %s",
-     db_name.ptr(), grantee.ptr(), tbl_name.ptr(), col_name.ptr(),
-     priv_type.ptr()));
+/**
+  Find all base tables or views of a view.
+*/
 
-    /*
-      Include grants for only users that exist at time of backup.
-    */
-    if (check_user_existence(m_thd, &grantee))
-      return new ColGrantObj(&grantee, &db_name, &tbl_name,
-                             &col_name, &priv_type);
-    else
-      return NULL;
-  }
-  else
-    return NULL;
-}
+class Find_view_underlying_tables: public Server_runnable
+{
+public:
+  Find_view_underlying_tables(const View_base_obj_iterator *base_obj_it,
+                              HASH *table_names,
+                              const String *db_name,
+                              const String *view_name);
 
-///////////////////////////////////////////////////////////////////////////
+  bool execute_server_code(THD *thd);
+private:
+  /* Store the resulting unique here */
+  const View_base_obj_iterator *m_base_obj_it;
+  HASH *m_table_names;
+  const String *m_db_name;
+  const String *m_view_name;
+};
 
-//
-// Implementation: ViewBaseObjectsIterator class.
-//
 
-///////////////////////////////////////////////////////////////////////////
+Find_view_underlying_tables::
+Find_view_underlying_tables(const View_base_obj_iterator *base_obj_it,
+                            HASH *table_names,
+                            const String *db_name,
+                            const String *view_name)
+  :m_base_obj_it(base_obj_it),
+   m_table_names(table_names),
+   m_db_name(db_name),
+   m_view_name(view_name)
+{
+}
 
-ViewBaseObjectsIterator *
-ViewBaseObjectsIterator::create(THD *thd,
-                               const String *db_name,
-                               const String *view_name,
-                               IteratorType iterator_type)
+
+bool
+Find_view_underlying_tables::execute_server_code(THD *thd)
 {
-  uint table_count; // Passed to open_tables(). Not used.
-  THD *my_thd= new THD();
+  bool res= TRUE;
+  uint counter_not_used; /* Passed to open_tables(). Not used. */
 
-  my_thd->security_ctx= thd->security_ctx;
+  TABLE_LIST *table_list;
 
-  my_thd->thread_stack= (char*) &my_thd;
-  my_thd->store_globals();
-  lex_start(my_thd);
-
-  TABLE_LIST *tl =
-    sp_add_to_query_tables(my_thd,
-                           my_thd->lex,
-                           ((String *) db_name)->c_ptr_safe(),
-                           ((String *) view_name)->c_ptr_safe(),
-                           TL_READ);
-
-  if (open_tables(my_thd, &tl, &table_count, 0))
-  {
-    close_thread_tables(my_thd);
-    delete my_thd;
-    thd->store_globals();
+  DBUG_ENTER("Find_view_underlying_tables::execute_server_code");
 
-    return NULL;
-  }
+  lex_start(thd);
+  table_list= sp_add_to_query_tables(thd, thd->lex,
+                                     ((String *) m_db_name)->c_ptr_safe(),
+                                     ((String *) m_view_name)->c_ptr_safe(),
+                                     TL_READ);
 
-  HASH *table_names = new HASH();
+  if (table_list == NULL)                      /* out of memory, reported */
+  {
+    lex_end(thd->lex);
+    DBUG_RETURN(TRUE);
+  }
 
-  hash_init(table_names, system_charset_info, 16, 0, 0,
-            get_table_name_key,
-            delete_table_name_key,
-            MYF(0));
+  if (open_tables(thd, &table_list, &counter_not_used,
+                  MYSQL_OPEN_SKIP_TEMPORARY))
+    goto end;
 
-  if (tl->view_tables)
+  if (table_list->view_tables)
   {
-    List_iterator_fast<TABLE_LIST> it(*tl->view_tables);
-    TABLE_LIST *tl2;
+    TABLE_LIST *table;
+    List_iterator_fast<TABLE_LIST> it(*table_list->view_tables);
 
-    while ((tl2 = it++))
+    /*
+      Iterate over immediate underlying tables.
+      The list doesn't include views or tables referenced indirectly,
+      through other views, or stored functions or triggers.
+    */
+    while ((table= it++))
     {
-      Table_name_key *tnk=
-        new Table_name_key(tl2->db, tl2->db_length,
-                           tl2->table_name, tl2->table_name_length);
+      Table_name_key *table_name_key;
+      char *key_buff;
 
-      if (iterator_type == GET_BASE_TABLES && tl2->view ||
-          iterator_type == GET_BASE_VIEWS && !tl2->view)
-      {
-        delete tnk;
+      /* If we expect a view, and it's a table, or vice versa, continue */
+      if ((int) m_base_obj_it->get_base_obj_kind() != test(table->view))
         continue;
-      }
 
-      if (!hash_search(table_names,
-                       (uchar *) tnk->key.c_ptr_safe(),
-                       tnk->key.length()))
-      {
-        my_hash_insert(table_names, (uchar *) tnk);
-      }
-      else
+      if (! my_multi_malloc(MYF(MY_WME),
+                            &table_name_key, sizeof(*table_name_key),
+                            &key_buff, table->mdl_lock_data->key_length,
+                            NullS))
+        goto end;
+
+      table_name_key->init_from_mdl_key(table->mdl_lock_data->key,
+                                        table->mdl_lock_data->key_length,
+                                        key_buff);
+
+      if (my_hash_insert(m_table_names, (uchar*) table_name_key))
       {
-        delete tnk;
+        my_free(table_name_key, MYF(0));
+        goto end;
       }
     }
   }
+  res= FALSE;
+  my_ok(thd);
 
-  close_thread_tables(my_thd);
-  delete my_thd;
-
-  thd->store_globals();
-
-  return new ViewBaseObjectsIterator(table_names);
+end:
+  lex_end(thd->lex);
+  DBUG_RETURN(res);
 }
 
-ViewBaseObjectsIterator::ViewBaseObjectsIterator(HASH *table_names) :
-  m_table_names(table_names),
-  m_cur_idx(0)
-{
-}
+///////////////////////////////////////////////////////////////////////////
 
-ViewBaseObjectsIterator::~ViewBaseObjectsIterator()
+bool View_base_obj_iterator::init(THD *thd,
+                                  const String *db_name,
+                                  const String *view_name)
 {
-  hash_free(m_table_names);
-  delete m_table_names;
-}
+  Find_view_underlying_tables find_tables(this, &m_table_names,
+                                          db_name, view_name);
+  Ed_result ed_result; /* Just to grab OK or ERROR */
 
-TableObj *ViewBaseObjectsIterator::next()
-{
-  if (m_cur_idx >= m_table_names->records)
-    return NULL;
+  if (mysql_execute_direct(thd, &find_tables, &ed_result))
+    return TRUE;
 
-  Table_name_key *tnk=
-    (Table_name_key *) hash_element(m_table_names, m_cur_idx);
+  /* The table list is filled with unique underlying table names. */
 
-  ++m_cur_idx;
-
-  return new TableObj(&tnk->db_name, &tnk->table_name, false);
+  return FALSE;
 }
 
 ///////////////////////////////////////////////////////////////////////////
 
-//
-// Implementation: enumeration functions.
-//
-
-///////////////////////////////////////////////////////////////////////////
-
-Obj_iterator *get_databases(THD *thd)
+Obj *View_base_obj_iterator::next()
 {
-  TABLE *is_table;
-  handler *ha;
-  my_bitmap_map *orig_columns;
-
-  if (InformationSchemaIterator::prepare_is_table(
-      thd, &is_table, &ha, &orig_columns, SCH_SCHEMATA,
-      thd->lex->db_list))
+  if (m_cur_idx >= m_table_names.records)
     return NULL;
 
-  return new DatabaseIterator(thd, is_table, ha, orig_columns);
-}
+  Table_name_key *table_name_key=
+    (Table_name_key *) hash_element(&m_table_names, m_cur_idx);
 
-template <typename Iterator>
-Iterator *create_is_iterator(THD *thd,
-                             enum_schema_tables is_table_idx,
-                             const String *db_name)
-{
-  TABLE *is_table;
-  handler *ha;
-  my_bitmap_map *orig_columns;
-
-  LEX_STRING dbname;
-  String db;
-  db.copy(*db_name);
-  thd->make_lex_string(&dbname, db.c_ptr(), db.length(), FALSE);
-  List<LEX_STRING> db_list;
-  db_list.push_back(&dbname);
-
-  if (InformationSchemaIterator::prepare_is_table(
-      thd, &is_table, &ha, &orig_columns, is_table_idx,
-      db_list))
-    return NULL;
+  ++m_cur_idx;
 
-  return new Iterator(thd, db_name, is_table, ha, orig_columns);
+  return create_obj(&table_name_key->db_name, &table_name_key->table_name);
 }
 
-template
-DbTablesIterator *
-create_is_iterator<DbTablesIterator>(THD *, enum_schema_tables, const String *);
-
-template
-DbViewsIterator *
-create_is_iterator<DbViewsIterator>(THD *, enum_schema_tables, const String *);
-
-template
-DbTriggerIterator *
-create_is_iterator<DbTriggerIterator>(THD *, enum_schema_tables, const String *);
-
-template
-DbStoredProcIterator *
-create_is_iterator<DbStoredProcIterator>(THD *, enum_schema_tables, const String *);
-
-template
-DbStoredFuncIterator *
-create_is_iterator<DbStoredFuncIterator>(THD *, enum_schema_tables, const String *);
-
-#ifdef HAVE_EVENT_SCHEDULER
-template
-DbEventIterator *
-create_is_iterator<DbEventIterator>(THD *, enum_schema_tables, const String *);
-#endif
-
-template
-DbGrantIterator *
-create_is_iterator<DbGrantIterator>(THD *, enum_schema_tables, const String *);
-
-template
-TblGrantIterator *
-create_is_iterator<TblGrantIterator>(THD *, enum_schema_tables, const String *);
-
-template
-ColGrantIterator *
-create_is_iterator<ColGrantIterator>(THD *, enum_schema_tables, const String *);
+///////////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////
 
-Obj_iterator *get_db_tables(THD *thd, const String *db_name)
-{
-  return create_is_iterator<DbTablesIterator>(thd, SCH_TABLES, db_name);
-}
+/**
+  @class View_base_table_iterator
 
-Obj_iterator *get_db_views(THD *thd, const String *db_name)
-{
-  return create_is_iterator<DbViewsIterator>(thd, SCH_TABLES, db_name);
-}
+  This is an iterator over base tables for a view.
+*/
 
-Obj_iterator *get_db_triggers(THD *thd, const String *db_name)
+class View_base_table_iterator : public View_base_obj_iterator
 {
-  return create_is_iterator<DbTriggerIterator>(thd, SCH_TRIGGERS, db_name);
-}
+public:
+  static inline View_base_obj_iterator *
+  create(THD *thd, const String *db_name, const String *view_name)
+  {
+    return View_base_obj_iterator::create<View_base_table_iterator>
+      (thd, db_name, view_name);
+  }
 
-Obj_iterator *get_db_stored_procedures(THD *thd, const String *db_name)
-{
-  return create_is_iterator<DbStoredProcIterator>(thd, SCH_PROCEDURES, db_name);
-}
+protected:
+  virtual inline enum_base_obj_kind get_base_obj_kind() const
+  { return BASE_TABLE; }
 
-Obj_iterator *get_db_stored_functions(THD *thd, const String *db_name)
-{
-  return create_is_iterator<DbStoredFuncIterator>(thd, SCH_PROCEDURES, db_name);
-}
+  virtual inline Obj *create_obj(const LEX_STRING *db_name,
+                                 const LEX_STRING *obj_name)
+  { return new Table_obj(*db_name, *obj_name); }
+};
 
-Obj_iterator *get_db_events(THD *thd, const String *db_name)
-{
-#ifdef HAVE_EVENT_SCHEDULER
-  return create_is_iterator<DbEventIterator>(thd, SCH_EVENTS, db_name);
-#else
-  return (Obj_iterator *)new ObjIteratorDummyImpl;
-#endif
-}
+///////////////////////////////////////////////////////////////////////////
 
-/**
-  GrantObjIterator constructor
+template
+View_base_table_iterator *
+View_base_obj_iterator::
+create<View_base_table_iterator>(THD *thd, const String *db_name,
+                                 const String *view_name);
 
-  This constructor initializes iterators for the grants supported.
-  These include database-, table- and routine-, and column-level grants.
-  The iterators return all of the grants for the database specified.
-*/
-GrantObjIterator::GrantObjIterator(THD *thd, const String *db_name)
-: Obj_iterator()
-{
-  db_grants= create_is_iterator<DbGrantIterator>(thd,
-                                                 SCH_SCHEMA_PRIVILEGES,
-                                                 db_name);
-  tbl_grants= create_is_iterator<TblGrantIterator>(thd,
-                                                 SCH_TABLE_PRIVILEGES,
-                                                 db_name);
-  col_grants= create_is_iterator<ColGrantIterator>(thd,
-                                                 SCH_COLUMN_PRIVILEGES,
-                                                 db_name);
-}
-
-Obj *GrantObjIterator::next()
-{
-  Obj *obj= 0;
-  obj= db_grants->next();
-  if (!obj)
-    obj= tbl_grants->next();
-  if (!obj)
-    obj= col_grants->next();
-  return obj;
-}
+///////////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////
 
 /**
-  Creates a high-level iterator that iterates over database-, table-,
-  routine-, and column-level privileges which shall permit a single
-  iterator from the si_objects to retrieve all of the privileges for
-  a given database.
-
-  @param[IN] thd      Current THD object
-  @param[IN] db_name  Name of database to get grants
+  @class View_base_view_iterator
 
-  @Note The client is responsible for destroying the returned iterator.
-
-  @return a pointer to an iterator object.
-    @retval NULL in case of error.
+  This is an iterator over base views for a view.
 */
-Obj_iterator *get_all_db_grants(THD *thd, const String *db_name)
+
+class View_base_view_iterator : public View_base_obj_iterator
 {
-  return new GrantObjIterator(thd, db_name);
-}
+public:
+  static inline View_base_obj_iterator *
+  create(THD *thd, const String *db_name, const String *view_name)
+  {
+    return View_base_obj_iterator::create<View_base_view_iterator>
+      (thd, db_name, view_name);
+  }
 
-///////////////////////////////////////////////////////////////////////////
+protected:
+  virtual inline enum_base_obj_kind get_base_obj_kind() const { return VIEW; }
 
-//
-// Implementation: dependency functions.
-//
+  virtual inline Obj *create_obj(const LEX_STRING *db_name,
+                                 const LEX_STRING *obj_name)
+  { return new View_obj(*db_name, *obj_name); }
+};
 
 ///////////////////////////////////////////////////////////////////////////
 
-Obj_iterator* get_view_base_tables(THD *thd,
-                                   const String *db_name,
-                                   const String *view_name)
-{
-  return ViewBaseObjectsIterator::create(
-    thd, db_name, view_name, ViewBaseObjectsIterator::GET_BASE_TABLES);
-}
-
-Obj_iterator* get_view_base_views(THD *thd,
-                                  const String *db_name,
-                                  const String *view_name)
-{
-  return ViewBaseObjectsIterator::create(
-    thd, db_name, view_name, ViewBaseObjectsIterator::GET_BASE_VIEWS);
-}
+template
+View_base_view_iterator *
+View_base_obj_iterator::
+create<View_base_view_iterator>(THD *thd, const String *db_name,
+                                const String *view_name);
 
 ///////////////////////////////////////////////////////////////////////////
-
-//
-// Implementation: DatabaseObj class.
-//
-
 ///////////////////////////////////////////////////////////////////////////
 
-DatabaseObj::DatabaseObj(const String *db_name)
-{
-  m_db_name.copy(*db_name); // copy name string to newly allocated memory
-}
-
 /**
   Serialize the object.
 
   This method produces the data necessary for materializing the object
   on restore (creates object).
 
-  @param[in]  thd            Thread context.
-  @param[out] serialization  The data needed to recreate this object.
+  @param[in]  thd         Thread context.
+  @param[out] out_stream  Output stream.
 
   @note this method will return an error if the db_name is either
-        mysql or information_schema as these are not objects that
-        should be recreated using this interface.
+  mysql or information_schema as these are not objects that
+  should be recreated using this interface.
 
   @returns Error status.
-   @retval FALSE on success
-   @retval TRUE on error
+    @retval FALSE on success.
+    @retval TRUE on error.
 */
-bool DatabaseObj::do_serialize(THD *thd, String *serialization)
+
+bool Database_obj::do_serialize(THD *thd, Out_stream &out_stream)
 {
-  HA_CREATE_INFO create;
-  DBUG_ENTER("DatabaseObj::serialize()");
-  DBUG_PRINT("DatabaseObj::serialize", ("name: %s", m_db_name.c_ptr()));
+  DBUG_ENTER("Database_obj::do_serialize");
+  DBUG_PRINT("Database_obj::do_serialize",
+             ("name: %.*s", STR(*get_name())));
 
-  if (is_internal_db_name(&m_db_name))
+  if (is_internal_db_name(get_name()))
   {
-    DBUG_PRINT("backup",(" Skipping internal database %s", m_db_name.c_ptr()));
-    DBUG_RETURN(TRUE);
-  }
-  create.default_table_charset= system_charset_info;
+    DBUG_PRINT("backup",
+               (" Skipping internal database %.*s", STR(*get_name())));
 
-  if (check_db_dir_existence(m_db_name.c_ptr()))
-  {
-    my_error(ER_BAD_DB_ERROR, MYF(0), m_db_name.c_ptr());
     DBUG_RETURN(TRUE);
   }
 
-  load_db_opt_by_name(thd, m_db_name.c_ptr(), &create);
+  /* Run 'SHOW CREATE' query. */
 
-  serialization->append(STRING_WITH_LEN("CREATE DATABASE "));
-  append_identifier(thd, serialization, m_db_name.c_ptr(), m_db_name.length());
+  Ed_result ed_result;
 
-  if (create.default_table_charset)
   {
-    serialization->append(STRING_WITH_LEN(" DEFAULT CHARACTER SET "));
-    serialization->append(create.default_table_charset->csname);
-    if (!(create.default_table_charset->state & MY_CS_PRIMARY))
+    String_stream s_stream;
+    s_stream <<
+      "SHOW CREATE DATABASE `" << get_name() << "`";
+
+    if (run_service_interface_sql(thd, s_stream.lex_string(), &ed_result) ||
+        ed_result.get_warnings().elements > 0)
     {
-      serialization->append(STRING_WITH_LEN(" COLLATE "));
-      serialization->append(create.default_table_charset->name);
+      /*
+        There should be no warnings. A warning means that serialization has
+        failed.
+      */
+      DBUG_RETURN(TRUE);
     }
   }
-  DBUG_RETURN(FALSE);
-}
-
-/**
-  Materialize the serialization string.
 
-  This method saves serialization string into a member variable.
+  /* Check result. */
 
-  @param[in]  serialization_version   version number of this interface
-  @param[in]  serialization           the string from serialize()
+  /* The result must contain only one result-set... */
+  DBUG_ASSERT(ed_result.elements == 1);
 
-  @todo take serialization_version into account
+  Ed_result_set *ed_result_set= ed_result.get_cur_result_set();
 
-  @returns Error status.
-    @retval FALSE on success
-    @retval TRUE on error
-  */
-bool DatabaseObj::materialize(uint serialization_version,
-                             const String *serialization)
-{
-  DBUG_ENTER("DatabaseObj::materialize()");
-  m_create_stmt.copy(*serialization);
-  DBUG_RETURN(FALSE);
-}
+  /* ... which is not NULL. */
+  DBUG_ASSERT(ed_result_set);
 
-/**
-  Create the object.
+  if (ed_result_set->data()->elements == 0)
+    DBUG_RETURN(TRUE);
 
-  This method uses serialization string in a query and executes it.
+  /* There must be one row. */
+  DBUG_ASSERT(ed_result_set->data()->elements == 1);
 
-  @param[in]  thd  Thread context.
+  List_iterator_fast<Ed_row> row_it(*ed_result_set->data());
+  Ed_row *row= row_it++;
 
-  @returns Error status.
-    @retval FALSE on success
-    @retval TRUE on error
-*/
-bool DatabaseObj::do_execute(THD *thd)
-{
-  DBUG_ENTER("DatabaseObj::execute()");
-  drop(thd);
-  DBUG_RETURN(silent_exec(thd, &m_create_stmt));
-}
+  /* There must be two columns: database name and create statement. */
+  DBUG_ASSERT(row->get_metadata()->get_num_columns() == 2);
 
-/**
-  Drop the object.
-
-  This method calls the silent_exec method to execute the query.
-
-  @note This uses "IF EXISTS" and does not return error if
-        object does not exist.
+  /* Generate image. */
 
-        @param[in]  thd            Thread context.
-  @param[out] serialization  the data needed to recreate this object
+  out_stream << row->get_column(1);
 
-  @returns Error status.
-    @retval FALSE on success
-    @retval TRUE on error
-*/
-bool DatabaseObj::drop(THD *thd)
-{
-  DBUG_ENTER("DatabaseObj::drop()");
-  DBUG_RETURN(drop_object(thd,
-                          (char *) "DATABASE",
-                          0,
-                          &m_db_name));
+  DBUG_RETURN(FALSE);
 }
 
 ///////////////////////////////////////////////////////////////////////////
 
-//
-// Implementation: TableObj class.
-//
-
-///////////////////////////////////////////////////////////////////////////
-
-TableObj::TableObj(const String *db_name,
-                   const String *table_name,
-                   bool table_is_view) :
-  m_table_is_view(table_is_view)
+void Database_obj::build_drop_statement(String_stream &s_stream) const
 {
-  m_db_name.copy(*db_name);
-  m_table_name.copy(*table_name);
+  s_stream <<
+    "DROP DATABASE IF EXISTS `" << get_name() << "`";
 }
 
-bool TableObj::serialize_table(THD *thd, String *serialization)
-{
-  return 0;
-}
-
-bool TableObj::serialize_view(THD *thd, String *serialization)
-{
-  return 0;
-}
+///////////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////
 
 /**
   Serialize the object.
@@ -2083,1225 +1743,883 @@ bool TableObj::serialize_view(THD *thd, 
   This method produces the data necessary for materializing the object
   on restore (creates object).
 
-  @param[in]  thd            Thread context.
-  @param[out] serialization  The data needed to recreate this object.
+  @param[in]  thd         Thread context.
+  @param[out] out_stream  Output stream.
 
   @returns Error status.
-    @retval FALSE on success
-    @retval TRUE on error
+    @retval FALSE on success.
+    @retval TRUE on error.
 */
-bool TableObj::do_serialize(THD *thd, String *serialization)
-{
-  bool ret= 0;
-  LEX_STRING tname, dbname;
-  DBUG_ENTER("TableObj::serialize()");
-  DBUG_PRINT("TableObj::serialize", ("name: %s@%s", m_db_name.c_ptr(),
-             m_table_name.c_ptr()));
-
-  prepend_db(thd, serialization, &m_db_name);
-  tname.str= m_table_name.c_ptr();
-  tname.length= m_table_name.length();
-  dbname.str= m_db_name.c_ptr();
-  dbname.length= m_db_name.length();
-  Table_ident *name_id= new Table_ident(tname);
-  name_id->db= dbname;
 
-  /*
-    Add the view to the table list and set the thd to look at views only.
-    Note: derived from sql_yacc.yy.
-  */
-  thd->lex->select_lex.add_table_to_list(thd, name_id, NULL, 0);
-  TABLE_LIST *table_list= (TABLE_LIST*)thd->lex->select_lex.table_list.first;
-  thd->lex->sql_command = SQLCOM_SHOW_CREATE;
+bool Table_obj::do_serialize(THD *thd, Out_stream &out_stream)
+{
+  DBUG_ENTER("Table_obj::do_serialize");
+  DBUG_PRINT("Table_obj::do_serialize",
+             ("name: %.*s.%.*s",
+              STR(m_db_name), STR(m_id)));
 
-  /*
-    Setup view specific variables and settings
-  */
-  if (m_table_is_view)
-  {
-    thd->lex->only_view= 1;
-    thd->lex->view_prepare_mode= TRUE; // use prepare mode
-    table_list->skip_temporary= 1;     // skip temporary tables
-  }
+  Ed_result ed_result;
+  String_stream s_stream;
 
-  /*
-    Open the view and its base tables or views
-  */
-  if (open_normal_and_derived_tables(thd, table_list, 0)) {
-    close_thread_tables(thd);
-    thd->lex->select_lex.table_list.empty();
-    DBUG_RETURN(TRUE);
-  }
+  s_stream <<
+    "SHOW CREATE TABLE `" << &m_db_name << "`.`" << &m_id << "`";
 
-  /*
-    Setup view specific variables and settings
-  */
-  if (m_table_is_view)
+  if (run_service_interface_sql(thd, s_stream.lex_string(), &ed_result) ||
+      ed_result.get_warnings().elements > 0)
   {
-    View_creation_ctx *creation_ctx= table_list->view_creation_ctx;
-
-    /*
-      append character set client charset information
-    */
-    serialization->append("SET CHARACTER_SET_CLIENT = '");
-    serialization->append(creation_ctx->get_client_cs()->csname);
-    serialization->append("'; ");
-
     /*
-      append collation_connection information
+      There should be no warnings. A warning means that serialization has
+      failed.
     */
-    serialization->append("SET COLLATION_CONNECTION = '");
-    serialization->append(creation_ctx->get_connection_cl()->name);
-    serialization->append("'; ");
-
-    table_list->view_db= dbname;
-    serialization->set_charset(creation_ctx->get_client_cs());
+    DBUG_RETURN(TRUE);
   }
 
-  /*
-    Get the create statement and close up shop.
-  */
-  ret= m_table_is_view ?
-    view_store_create_info(thd, table_list, serialization) :
-    store_create_info(thd, table_list, serialization, NULL,
-                      /* show_database */ TRUE);
-  close_thread_tables(thd);
-  serialization->set_charset(system_charset_info);
-  thd->lex->select_lex.table_list.empty();
-  DBUG_RETURN(FALSE);
-}
-
-/**
-  Materialize the serialization string.
+  /* Check result. */
 
-  This method saves serialization string into a member variable.
+  /* The result must contain only one result-set... */
+  DBUG_ASSERT(ed_result.elements == 1);
 
-  @param[in]  serialization_version   version number of this interface
-  @param[in]  serialization           the string from serialize()
-
-  @todo take serialization_version into account
-
-  @returns Error status.
-    @retval FALSE on success
-    @retval TRUE on error
-  */
-bool TableObj::materialize(uint serialization_version,
-                           const String *serialization)
-{
-  DBUG_ENTER("TableObj::materialize()");
-  m_create_stmt.copy(*serialization);
-  DBUG_RETURN(FALSE);
-}
+  Ed_result_set *ed_result_set= ed_result.get_cur_result_set();
 
-/**
-  Create the object represented by TableObj in the database.
+  /* ... which is not NULL. */
+  DBUG_ASSERT(ed_result_set);
 
-  This method uses serialization string in a query and executes it.
+  if (ed_result_set->data()->elements == 0)
+    DBUG_RETURN(TRUE);
 
-  @param[in]  thd  Thread context.
+  /* There must be one row. */
+  DBUG_ASSERT(ed_result_set->data()->elements == 1);
 
-  @returns Error status.
-    @retval FALSE on success
-    @retval TRUE on error
-*/
-bool TableObj::do_execute(THD *thd)
-{
-  DBUG_ENTER("TableObj::execute()");
-  drop(thd);
-  DBUG_RETURN(silent_exec(thd, &m_create_stmt));
-}
+  List_iterator_fast<Ed_row> row_it(*ed_result_set->data());
+  Ed_row *row= row_it++;
 
-/**
-  Drop the object.
+  /* There must be two columns: database name and create statement. */
+  DBUG_ASSERT(row->get_metadata()->get_num_columns() == 2);
 
-  This method calls the silent_exec method to execute the query.
+  /* Generate serialization image. */
 
-  @note This uses "IF EXISTS" and does not return error if
-        object does not exist.
+  {
+    s_stream.reset();
+    s_stream << "USE `" << &m_db_name << "`";
+    out_stream << s_stream;
+  }
 
-  @param[in]  thd            Thread context.
-  @param[out] serialization  The data needed to recreate this object.
+  out_stream << row->get_column(1);
 
-  @returns Error status.
-    @retval FALSE on success
-    @retval TRUE on error
-*/
-bool TableObj::drop(THD *thd)
-{
-  DBUG_ENTER("TableObj::drop()");
-  DBUG_RETURN(drop_object(thd,
-                          (char *) "TABLE",
-                          &m_db_name,
-                          &m_table_name));
+  DBUG_RETURN(FALSE);
 }
 
 ///////////////////////////////////////////////////////////////////////////
-
-//
-// Implementation: TriggerObj class.
-//
 ///////////////////////////////////////////////////////////////////////////
 
-TriggerObj::TriggerObj(const String *db_name,
-                             const String *trigger_name)
-{
-  // copy strings to newly allocated memory
-  m_db_name.copy(*db_name);
-  m_trigger_name.copy(*trigger_name);
-}
+static bool
+get_view_create_stmt(THD *thd,
+                     View_obj *view,
+                     LEX_STRING *create_stmt,
+                     LEX_STRING *client_cs_name,
+                     LEX_STRING *connection_cl_name)
+{
+  /* Get a create statement for a view. */
+  Ed_result ed_result;
+  String_stream s_stream;
+
+  s_stream <<
+    "SHOW CREATE VIEW `" << view->get_db_name() << "`."
+    "`" << view->get_name() << "`";
 
-/**
-  Serialize the object.
+  if (run_service_interface_sql(thd, s_stream.lex_string(), &ed_result) ||
+      ed_result.get_warnings().elements > 0)
+  {
+    /*
+      There should be no warnings. A warning means that serialization has
+      failed.
+    */
+    return TRUE;
+  }
 
-  This method produces the data necessary for materializing the object
-  on restore (creates object).
+  /* The result must contain only one result-set... */
+  DBUG_ASSERT(ed_result.elements == 1);
 
-  @param[in]  thd            Thread handler.
-  @param[out] serialization  The data needed to recreate this object.
+  Ed_result_set *ed_result_set= ed_result.get_cur_result_set();
 
-  @note this method will return an error if the db_name is either
-        mysql or information_schema as these are not objects that
-        should be recreated using this interface.
+  /* ... which is not NULL. */
+  DBUG_ASSERT(ed_result_set);
 
-  @returns Error status.
-    @retval FALSE on success
-    @retval TRUE on error
-*/
-bool TriggerObj::do_serialize(THD *thd, String *serialization)
-{
-  bool ret= false;
-  uint num_tables;
-  sp_name *trig_name;
-  LEX_STRING trg_name;
-  ulonglong trg_sql_mode;
-  LEX_STRING trg_sql_mode_str;
-  LEX_STRING trg_sql_original_stmt;
-  LEX_STRING trg_client_cs_name;
-  LEX_STRING trg_connection_cl_name;
-  LEX_STRING trg_db_cl_name;
-  CHARSET_INFO *trg_client_cs;
-  DBUG_ENTER("TriggerObj::serialize()");
-
-  DBUG_PRINT("TriggerObj::serialize", ("name: %s in %s",
-             m_trigger_name.c_ptr(), m_db_name.c_ptr()));
-
-  prepend_db(thd, serialization, &m_db_name);
-  LEX_STRING db;
-  db.str= m_db_name.c_ptr();
-  db.length= m_db_name.length();
-  LEX_STRING t_name;
-  t_name.str= m_trigger_name.c_ptr();
-  t_name.length= m_trigger_name.length();
-  trig_name= new sp_name(db, t_name, true);
-  trig_name->init_qname(thd);
-  TABLE_LIST *lst= get_trigger_table(thd, trig_name);
-  if (!lst)
-    DBUG_RETURN(FALSE);
+  if (ed_result_set->data()->elements == 0)
+    return TRUE;
 
-  alloc_mdl_locks(lst, thd->mem_root);
+  /* There must be one row. */
+  DBUG_ASSERT(ed_result_set->data()->elements == 1);
 
-  DBUG_EXECUTE_IF("backup_fail_add_trigger", DBUG_RETURN(TRUE););
-  if (open_tables(thd, &lst, &num_tables, 0))
-    DBUG_RETURN(TRUE);
+  List_iterator_fast<Ed_row> row_it(*ed_result_set->data());
+  Ed_row *row= row_it++;
 
-  DBUG_ASSERT(num_tables == 1);
-  Table_triggers_list *triggers= lst->table->triggers;
-  if (!triggers)
-    DBUG_RETURN(FALSE);
+  /* There must be four columns. */
+  DBUG_ASSERT(row->get_metadata()->get_num_columns() == 4);
 
-  int trigger_idx= triggers->find_trigger_by_name(&trig_name->m_name);
-  if (trigger_idx < 0)
-    DBUG_RETURN(FALSE);
+  const LEX_STRING *c1= row->get_column(1);
+  const LEX_STRING *c2= row->get_column(2);
+  const LEX_STRING *c3= row->get_column(3);
 
-  triggers->get_trigger_info(thd,
-                             trigger_idx,
-                             &trg_name,
-                             &trg_sql_mode,
-                             &trg_sql_original_stmt,
-                             &trg_client_cs_name,
-                             &trg_connection_cl_name,
-                             &trg_db_cl_name);
-  sys_var_thd_sql_mode::symbolic_mode_representation(thd,
-                                                     trg_sql_mode,
-                                                     &trg_sql_mode_str);
+  create_stmt->str= thd->strmake(c1->str, c1->length);
+  create_stmt->length= c1->length;
 
-  /*
-    prepend SQL Mode
-  */
-  serialization->append("SET SQL_MODE = '");
-  serialization->append(trg_sql_mode_str.str);
-  serialization->append("'; ");
+  client_cs_name->str= thd->strmake(c2->str, c2->length);
+  client_cs_name->length= c2->length;
 
-  /*
-    append character set client charset information
-  */
-  serialization->append("SET CHARACTER_SET_CLIENT = '");
-  serialization->append(trg_client_cs_name.str);
-  serialization->append("'; ");
+  connection_cl_name->str= thd->strmake(c3->str, c3->length);
+  connection_cl_name->length= c3->length;
 
-  /*
-    append collation_connection information
-  */
-  serialization->append("SET COLLATION_CONNECTION = '");
-  serialization->append(trg_connection_cl_name.str);
-  serialization->append("'; ");
-
-  /*
-    append collation_connection information
-  */
-  serialization->append("SET COLLATION_DATABASE = '");
-  serialization->append(trg_db_cl_name.str);
-  serialization->append("'; ");
-
-  if (resolve_charset(trg_client_cs_name.str, NULL, &trg_client_cs))
-    ret= false;
-  else
-    serialization->append(trg_sql_original_stmt.str);
-  close_thread_tables(thd);
-  thd->lex->select_lex.table_list.empty();
-  serialization->set_charset(system_charset_info);
-  DBUG_RETURN(ret);
+  return FALSE;
 }
 
-/**
-  Materialize the serialization string.
+///////////////////////////////////////////////////////////////////////////
 
-  This method saves serialization string into a member variable.
+/**
+  Serialize the object.
 
-  @param[in]  serialization_version   version number of this interface
-  @param[in]  serialization           the string from serialize()
+  This method produces the data necessary for materializing the object
+  on restore (creates object).
 
-  @todo take serialization_version into account
+  @param[in]  thd         Thread context.
+  @param[out] out_stream  Output stream.
 
   @returns Error status.
-    @retval FALSE on success
-    @retval TRUE on error
+    @retval FALSE on success.
+    @retval TRUE on error.
 */
-bool TriggerObj::materialize(uint serialization_version,
-                             const String *serialization)
+bool View_obj::do_serialize(THD *thd, Out_stream &out_stream)
 {
-  DBUG_ENTER("TriggerObj::materialize()");
-  m_create_stmt.copy(*serialization);
-  DBUG_RETURN(0);
-}
-
-/**
-  Create the object.
+  DBUG_ENTER("View_obj::do_serialize");
+  DBUG_PRINT("View_obj::do_serialize",
+             ("name: %.*s.%.*s",
+              STR(m_db_name), STR(m_id)));
 
-  This method uses serialization string in a query and executes it.
+  String_stream s_stream;
 
-  @param[in]  thd  Thread context.
-
-  @returns Error status.
-    @retval FALSE on success
-    @retval TRUE on error
-*/
-bool TriggerObj::do_execute(THD *thd)
-{
-  DBUG_ENTER("TriggerObj::execute()");
-  drop(thd);
-  DBUG_RETURN(execute_with_ctx(thd, &m_create_stmt, false));
-}
+  LEX_STRING create_stmt= null_lex_str;
+  LEX_STRING client_cs_name= null_lex_str;
+  LEX_STRING connection_cl_name= null_lex_str;
 
-/**
-  Drop the object.
+  if (get_view_create_stmt(thd, this, &create_stmt,
+                           &client_cs_name, &connection_cl_name))
+  {
+    DBUG_RETURN(TRUE);
+  }
 
-  This method calls the silent_exec method to execute the query.
+  s_stream << "USE `" << &m_db_name << "`";
+  out_stream << s_stream;
 
-  @note This uses "IF EXISTS" and does not return error if
-        object does not exist.
+  s_stream.reset();
+  s_stream << "SET character_set_client = " << &client_cs_name;
+  out_stream << s_stream;
+
+  s_stream.reset();
+  s_stream << "SET collation_connection = " << &connection_cl_name;
+  out_stream << s_stream;
 
-  @param[in]  thd            Thread context.
-  @param[out] serialization  The data needed to recreate this object.
+  out_stream << &create_stmt;
 
-  @returns Error status.
-    @retval FALSE on success
-    @retval TRUE on error
-*/
-bool TriggerObj::drop(THD *thd)
-{
-  DBUG_ENTER("TriggerObj::drop()");
-  DBUG_RETURN(drop_object(thd,
-                          (char *) "TRIGGER",
-                          &m_db_name,
-                          &m_trigger_name));
+  DBUG_RETURN(FALSE);
 }
 
 ///////////////////////////////////////////////////////////////////////////
-
-//
-// Implementation: StoredProcObj class.
-//
-
 ///////////////////////////////////////////////////////////////////////////
 
-StoredProcObj::StoredProcObj(const String *db_name,
-                             const String *stored_proc_name)
+bool Stored_program_obj::do_serialize(THD *thd, Out_stream &out_stream)
 {
-  // copy strings to newly allocated memory
-  m_db_name.copy(*db_name);
-  m_stored_proc_name.copy(*stored_proc_name);
-}
-
-/**
-  Serialize the object.
+  DBUG_ENTER("Stored_program_obj::do_serialize");
+  DBUG_PRINT("Stored_program_obj::do_serialize",
+             ("name: %.*s.%.*s",
+              STR(m_db_name), STR(m_id)));
 
-  This method produces the data necessary for materializing the object
-  on restore (creates object).
+  DBUG_EXECUTE_IF("backup_fail_add_trigger", DBUG_RETURN(TRUE););
 
-  @param[in]  thd            Thread context.
-  @param[out] serialization  The data needed to recreate this object.
+  String_stream s_stream;
+  Ed_result ed_result;
 
-  @note this method will return an error if the db_name is either
-        mysql or information_schema as these are not objects that
-        should be recreated using this interface.
+  s_stream <<
+    "SHOW CREATE " << get_type_name() <<
+    " `" << &m_db_name << "`.`" << &m_id << "`";
 
-  @returns Error status.
-*/
-bool StoredProcObj::do_serialize(THD *thd, String *serialization)
-{
-  bool ret= false;
-  DBUG_ENTER("StoredProcObj::serialize()");
-  DBUG_PRINT("StoredProcObj::serialize", ("name: %s in %s",
-             m_stored_proc_name.c_ptr(), m_db_name.c_ptr()));
-  prepend_db(thd, serialization, &m_db_name);
-  ret= serialize_routine(thd, TYPE_ENUM_PROCEDURE, m_db_name,
-                         m_stored_proc_name, serialization);
-  serialization->set_charset(system_charset_info);
-  DBUG_RETURN(ret);
-}
+  if (run_service_interface_sql(thd, s_stream.lex_string(), &ed_result) ||
+      ed_result.get_warnings().elements > 0)
+  {
+    /*
+      There should be no warnings. A warning means that serialization has
+      failed.
+    */
+    DBUG_RETURN(TRUE);
+  }
 
-/**
-  Materialize the serialization string.
+  /* The result must contain only one result-set... */
+  DBUG_ASSERT(ed_result.elements == 1);
 
-  This method saves serialization string into a member variable.
+  Ed_result_set *ed_result_set= ed_result.get_cur_result_set();
 
-  @param[in]  serialization_version   version number of this interface
-  @param[in]  serialization           the string from serialize()
+  /* ... which is not NULL. */
+  DBUG_ASSERT(ed_result_set);
 
-  @todo take serialization_version into account
+  if (ed_result_set->data()->elements == 0)
+    DBUG_RETURN(TRUE);
 
-  @returns Error status.
-    @retval FALSE on success
-    @retval TRUE on error
-*/
-bool StoredProcObj::materialize(uint serialization_version,
-                             const String *serialization)
-{
-  DBUG_ENTER("StoredProcObj::materialize()");
-  m_create_stmt.copy(*serialization);
-  DBUG_RETURN(0);
-}
+  /* There must be one row. */
+  DBUG_ASSERT(ed_result_set->data()->elements == 1);
 
-/**
-  Create the object.
+  List_iterator_fast<Ed_row> row_it(*ed_result_set->data());
+  Ed_row *row= row_it++;
 
-  This method uses serialization string in a query and executes it.
+  s_stream.reset();
+  s_stream << "USE `" << &m_db_name << "`";
+  out_stream << s_stream;
 
-  @param[in]  thd  current thread
+  dump_header(row, out_stream);
+  out_stream << get_create_stmt(row);
 
-  @returns Error status.
-    @retval FALSE on success
-    @retval TRUE on error
-*/
-bool StoredProcObj::do_execute(THD *thd)
-{
-  DBUG_ENTER("StoredProcObj::execute()");
-  drop(thd);
-  DBUG_RETURN(execute_with_ctx(thd, &m_create_stmt, false));
+  DBUG_RETURN(FALSE);
 }
 
-/**
-  Drop the object.
-
-  This method calls the silent_exec method to execute the query.
-
-  @note This uses "IF EXISTS" and does not return error if
-        object does not exist.
-
-  @param[in]  thd            current thread
-  @param[out] serialization  the data needed to recreate this object
+///////////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////
 
-  @returns Error status.
-    @retval FALSE on success
-    @retval TRUE on error
-*/
-bool StoredProcObj::drop(THD *thd)
+const LEX_STRING *Stored_routine_obj::get_create_stmt(Ed_row *row)
 {
-  DBUG_ENTER("StoredProcObj::drop()");
-  DBUG_RETURN(drop_object(thd,
-                          (char *) "PROCEDURE",
-                          &m_db_name,
-                          &m_stored_proc_name));
+  return row->get_column(2);
 }
 
 ///////////////////////////////////////////////////////////////////////////
 
-//
-// Implementation: StoredFuncObj class.
-//
-
-///////////////////////////////////////////////////////////////////////////
-
-StoredFuncObj::StoredFuncObj(const String *db_name,
-                             const String *stored_func_name)
+void Stored_routine_obj::dump_header(Ed_row *row, Out_stream &out_stream)
 {
-  // copy strings to newly allocated memory
-  m_db_name.copy(*db_name);
-  m_stored_func_name.copy(*stored_func_name);
-}
+  String_stream s_stream;
 
-/**
-  Serialize the object.
-
-  This method produces the data necessary for materializing the object
-  on restore (creates object).
+  s_stream <<
+    "SET character_set_client = " << row->get_column(3);
+  out_stream << s_stream;
 
-  @param[in]  thd            Thread context.
-  @param[out] serialization  The data needed to recreate this object.
+  s_stream.reset();
+  s_stream <<
+    "SET collation_connection = " << row->get_column(4);
+  out_stream << s_stream;
 
-  @note this method will return an error if the db_name is either
-        mysql or information_schema as these are not objects that
-        should be recreated using this interface.
+  s_stream.reset();
+  s_stream <<
+    "SET collation_database = " << row->get_column(5);
+  out_stream << s_stream;
 
-  @returns Error status.
-    @retval FALSE on success
-    @retval TRUE on error
- */
-bool  StoredFuncObj::do_serialize(THD *thd, String *serialization)
-{
-  bool ret= false;
-  DBUG_ENTER("StoredFuncObj::serialize()");
-  DBUG_PRINT("StoredProcObj::serialize", ("name: %s in %s",
-              m_stored_func_name.c_ptr(), m_db_name.c_ptr()));
-  prepend_db(thd, serialization, &m_db_name);
-  ret= serialize_routine(thd, TYPE_ENUM_FUNCTION, m_db_name,
-                         m_stored_func_name, serialization);
-  serialization->set_charset(system_charset_info);
-  DBUG_RETURN(ret);
+  s_stream.reset();
+  s_stream <<
+    "SET sql_mode = '" << row->get_column(1) << "'";
+  out_stream << s_stream;
 }
 
-/**
-  Materialize the serialization string.
-
-  This method saves serialization string into a member variable.
-
-  @param[in]  serialization_version   version number of this interface
-  @param[in]  serialization           the string from serialize()
+///////////////////////////////////////////////////////////////////////////
+/////////////////////////////////////////////////////////////////////////////
 
-  @todo take serialization_version into account
+#ifdef HAVE_EVENT_SCHEDULER
 
-  @returns Error status.
-    @retval FALSE on success
-    @retval TRUE on error
-*/
-bool StoredFuncObj::materialize(uint serialization_version,
-                             const String *serialization)
+void Event_obj::dump_header(Ed_row *row, Out_stream &out_stream)
 {
-  DBUG_ENTER("StoredFuncObj::materialize()");
-  m_create_stmt.copy(*serialization);
-  DBUG_RETURN(0);
-}
+  String_stream s_stream;
 
-/**
-  Create the object.
+  s_stream <<
+    "SET character_set_client = " << row->get_column(4);
+  out_stream << s_stream;
+
+  s_stream.reset();
+  s_stream <<
+    "SET collation_connection = " << row->get_column(5);
+  out_stream << s_stream;
+
+  s_stream.reset();
+  s_stream <<
+    "SET collation_database = " << row->get_column(6);
+  out_stream << s_stream;
+
+  s_stream.reset();
+  s_stream <<
+    "SET sql_mode = '" << row->get_column(1) << "'";
+  out_stream << s_stream;
+
+  s_stream.reset();
+  s_stream <<
+    "SET time_zone = '" << row->get_column(2) << "'";
+  out_stream << s_stream;
+}
 
-  This method uses serialization string in a query and executes it.
+#endif // HAVE_EVENT_SCHEDULER
 
-  @param[in]  thd  Thread context.
+///////////////////////////////////////////////////////////////////////////
+/////////////////////////////////////////////////////////////////////////////
 
-  @returns Error status.
-    @retval FALSE on success
-    @retval TRUE on error
-*/
-bool StoredFuncObj::do_execute(THD *thd)
+Tablespace_obj::
+Tablespace_obj(LEX_STRING ts_name,
+               LEX_STRING comment,
+               LEX_STRING data_file_name,
+               LEX_STRING engine)
+  : Abstract_obj(ts_name)
 {
-  DBUG_ENTER("StoredFuncObj::execute()");
-  drop(thd);
-  DBUG_RETURN(execute_with_ctx(thd, &m_create_stmt, false));
-}
+  m_comment.copy(comment.str, comment.length, system_charset_info);
+  m_data_file_name.copy(data_file_name.str, data_file_name.length,
+                        system_charset_info);
+  m_engine.copy(engine.str, engine.length, system_charset_info);
 
-/**
-  Drop the object.
-
-  This method calls the silent_exec method to execute the query.
-
-  @note This uses "IF EXISTS" and does not return error if
-        object does not exist.
+  m_description.length(0);
+}
 
-  @param[in]  thd            Thread context.
-  @param[out] serialization  The data needed to recreate this object.
 
-  @returns Error status.
-    @retval FALSE on success
-    @retval TRUE on error
-*/
-bool StoredFuncObj::drop(THD *thd)
+Tablespace_obj::Tablespace_obj(LEX_STRING ts_name)
+  : Abstract_obj(ts_name)
 {
-  DBUG_ENTER("StoredFuncObj::drop()");
-  DBUG_RETURN(drop_object(thd,
-                          (char *) "FUNCTION",
-                          &m_db_name,
-                          &m_stored_func_name));
+  m_comment.length(0);
+  m_data_file_name.length(0);
+  m_engine.length(0);
+
+  m_description.length(0);
 }
 
 ///////////////////////////////////////////////////////////////////////////
 
-//
-// Implementation: EventObj class.
-//
-
-/////////////////////////////////////////////////////////////////////////////
-#ifdef HAVE_EVENT_SCHEDULER
-EventObj::EventObj(const String *db_name,
-                   const String *event_name)
-{
-  // copy strings to newly allocated memory
-  m_db_name.copy(*db_name);
-  m_event_name.copy(*event_name);
-}
-
 /**
   Serialize the object.
 
   This method produces the data necessary for materializing the object
   on restore (creates object).
 
-  @param[in]  thd            Thread context.
-  @param[out] serialization  The data needed to recreate this object.
-
-  @note this method will return an error if the db_name is either
-        mysql or information_schema as these are not objects that
-        should be recreated using this interface.
+  @param[in]  thd Thread context.
+  @param[out] out_stream  Output stream.
 
   @returns Error status.
-    @retval FALSE on success
-    @retval TRUE on error
+    @retval FALSE on success.
+    @retval TRUE on error.
 */
-bool EventObj::do_serialize(THD *thd, String *serialization)
-{
-  bool ret= false;
-  Open_tables_state open_tables_backup;
-  Event_timed et;
-  LEX_STRING sql_mode;
-  DBUG_ENTER("EventObj::serialize()");
-  DBUG_PRINT("EventObj::serialize", ("name: %s.%s", m_db_name.c_ptr(),
-             m_event_name.c_ptr()));
-
-  prepend_db(thd, serialization, &m_db_name);
-  Event_db_repository *db_repository= Events::get_db_repository();
-  thd->reset_n_backup_open_tables_state(&open_tables_backup);
-  LEX_STRING db;
-  db.str= m_db_name.c_ptr();
-  db.length= m_db_name.length();
-  LEX_STRING ev;
-  ev.str= m_event_name.c_ptr();
-  ev.length= m_event_name.length();
-  ret= db_repository->load_named_event(thd, db, ev, &et);
-  thd->restore_backup_open_tables_state(&open_tables_backup);
-  if (sys_var_thd_sql_mode::symbolic_mode_representation(thd,
-    et.sql_mode, &sql_mode))
-    DBUG_RETURN(TRUE);
-  if (!ret)
-  {
-    /*
-      Prepend sql_mode command.
-    */
-    serialization->append("SET SQL_MODE = '");
-    serialization->append(sql_mode.str);
-    serialization->append("'; ");
-
-    /*
-      append time zone information
-    */
-    serialization->append("SET TIME_ZONE = '");
-    const String *tz= et.time_zone->get_name();
-    serialization->append(tz->ptr());
-    serialization->append("'; ");
-
-    /*
-      append character set client charset information
-    */
-    serialization->append("SET CHARACTER_SET_CLIENT = '");
-    serialization->append(et.creation_ctx->get_client_cs()->csname);
-    serialization->append("'; ");
 
-    /*
-      append collation_connection information
-    */
-    serialization->append("SET COLLATION_CONNECTION = '");
-    serialization->append(et.creation_ctx->get_connection_cl()->name);
-    serialization->append("'; ");
+bool Tablespace_obj::do_serialize(THD *thd, Out_stream &out_stream)
+{
+  DBUG_ENTER("Tablespace_obj::do_serialize");
 
-    /*
-      append collation_connection information
-    */
-    serialization->append("SET COLLATION_DATABASE = '");
-    serialization->append(et.creation_ctx->get_db_cl()->name);
-    serialization->append("'; ");
+  out_stream << get_description();
 
-    if (et.get_create_event(thd, serialization))
-      DBUG_RETURN(0);
-  }
-  serialization->set_charset(system_charset_info);
-  DBUG_RETURN(0);
+  DBUG_RETURN(FALSE);
 }
 
-/**
-  Materialize the serialization string.
+///////////////////////////////////////////////////////////////////////////
 
-  This method saves serialization string into a member variable.
+bool Tablespace_obj::init_from_image(uint image_version, const String *image)
+{
+  if (Abstract_obj::init_from_image(image_version, image))
+    return TRUE;
 
-  @param[in]  serialization_version   version number of this interface
-  @param[in]  serialization           the string from serialize()
+  List_iterator_fast<LEX_STRING> it(m_stmt_list);
+  LEX_STRING *desc= it++;
 
-  @todo take serialization_version into account
+  /* Tablespace description must not be NULL. */
+  DBUG_ASSERT(desc);
 
-  @returns Error status.
-    @retval FALSE on success
-    @retval TRUE on error
-*/
-bool EventObj::materialize(uint serialization_version,
-                             const String *serialization)
-{
-  DBUG_ENTER("EventObj::materialize()");
-  m_create_stmt.copy(*serialization);
-  DBUG_RETURN(0);
+  m_description.copy(desc->str, desc->length, system_charset_info);
+
+  return FALSE;
 }
 
-/**
-  Create the object.
+///////////////////////////////////////////////////////////////////////////
 
-  This method uses serialization string in a query and executes it.
+/**
+  Get a description of the tablespace object.
 
-  @param[in]  thd  Thread context.
+  This method returns the description of the object which is currently
+  the serialization image.
 
-  @returns Error status.
-    @retval FALSE on success
-    @retval TRUE on error
+  @returns Serialization string.
 */
-bool EventObj::do_execute(THD *thd)
+
+const String *Tablespace_obj::get_description()
 {
-  DBUG_ENTER("EventObj::execute()");
-  drop(thd);
-  DBUG_RETURN(execute_with_ctx(thd, &m_create_stmt, true));
-}
+  DBUG_ENTER("Tablespace_obj::get_description");
 
-/**
-  Drop the object.
+  /* Either description or id and data file name must be not empty. */
+  DBUG_ASSERT(m_description.length() ||
+              m_id.length() && m_data_file_name.length());
 
-  This method calls the silent_exec method to execute the query.
+  if (m_description.length())
+    DBUG_RETURN(&m_description);
 
-  @note This uses "IF EXISTS" and does not return error if
-        object does not exist.
+  /* Construct the CREATE TABLESPACE command from the variables. */
 
-  @param[in]  thd            Thread context.
-  @param[out] serialization  the data needed to recreate this object
+  m_description.length(0);
 
-  @returns Error status.
-    @retval FALSE on success
-    @retval TRUE on error
-*/
-bool EventObj::drop(THD *thd)
-{
-  DBUG_ENTER("EventObj::drop()");
-  DBUG_RETURN(drop_object(thd,
-                          (char *) "EVENT",
-                          &m_db_name,
-                          &m_event_name));
-}
-#endif // HAVE_EVENT_SCHEDULER
+  String_stream s_stream(&m_description);
 
-///////////////////////////////////////////////////////////////////////////
+  s_stream <<
+    "CREATE TABLESPACE `" << &m_id << "` "
+    "ADD DATAFILE '" << &m_data_file_name << "' ";
 
-//
-// Implementation: TablespaceObj class.
-//
+  if (m_comment.length())
+    s_stream << "COMMENT = '" << &m_comment << "' ";
 
-/////////////////////////////////////////////////////////////////////////////
+  s_stream << "ENGINE = " << &m_engine;
 
-TablespaceObj::TablespaceObj(const String *ts_name)
-{
-  // copy strings to newly allocated memory
-  m_ts_name.copy(*ts_name);
-  m_datafile.length(0);
-  m_comments.length(0);
-  m_engine.length(0);
+  DBUG_RETURN(&m_description);
 }
 
-/**
-  Serialize the object.
+///////////////////////////////////////////////////////////////////////////
 
-  This method produces the data necessary for materializing the object
-  on restore (creates object).
+void Tablespace_obj::build_drop_statement(String_stream &s_stream) const
+{
+  s_stream <<
+    "DROP TABLESPACE `" << &m_id << "` ENGINE = " << &m_engine;
+}
 
-  @param[in]  thd            Thread context.
-  @param[out] serialization  The data needed to recreate this object.
+///////////////////////////////////////////////////////////////////////////
+/////////////////////////////////////////////////////////////////////////////
 
-  @returns Error status.
-    @retval FALSE on success
-    @retval TRUE on error
-*/
-bool TablespaceObj::do_serialize(THD *thd, String *serialization)
+void
+Grant_obj::
+generate_unique_grant_id(const String *user_name, String *id)
 {
-  DBUG_ENTER("TablespaceObj::serialize()");
-  build_serialization();
-  serialization->copy(m_create_stmt);
-  DBUG_RETURN(FALSE);
-}
+  /*
+    @note This code is not MT-safe, but we don't need MT-safety here.
+    Backup has object cache (a hash) for one BACKUP/RESTORE statement.
+    Hash keys are formed from object names (Obj::get_db_name() and
+    Obj::get_name()). So, here unique keys for the object cache are
+    generated. These keys need to be unique only within one backup session
+    (serialization image).
+  */
 
-/**
-  Materialize the serialization string.
+  static unsigned long id_counter= 0;
 
-  This method saves serialization string into a member variable.
-  Also extracts tablespace engine name from serialization string.
+  id->length(0);
 
-  @param[in]  serialization_version   version number of this interface
-  @param[in]  serialization           the string from serialize()
+  String_stream s_stream(id);
 
-  @todo take serialization_version into account
+  if (user_name->length())
+    s_stream << "<empty>";
+  else
+    s_stream << user_name;
 
-  @returns Error status.
-    @retval FALSE on success
-    @retval TRUE on error
-*/
-bool TablespaceObj::materialize(uint serialization_version,
-                                const String *serialization)
-{
-  DBUG_ENTER("TablespaceObj::materialize()");
-  m_create_stmt.copy(*serialization);
-  /* Extract engine from create statement */
-  String tmp_str= String("=", 1, system_charset_info);
-  int pos= m_create_stmt.strrstr(tmp_str, m_create_stmt.length());
-  m_engine.copy(m_create_stmt.ptr() + pos + 1, m_create_stmt.length() - pos - 1,
-                system_charset_info);
-  DBUG_RETURN(FALSE);
+  s_stream << " " << Int_value(++id_counter);
 }
 
-/**
-  Get a description of the tablespace object.
-
-  This method returns the description of the object which is currently
-  the serialization string.
+/////////////////////////////////////////////////////////////////////////////
 
-  @returns Serialization string.
-*/
-const String *TablespaceObj::describe()
+Grant_obj::Grant_obj(const Ed_row &row)
+  : Abstract_obj(null_lex_str)
 {
-  DBUG_ENTER("TablespaceObj::describe()");
-  DBUG_RETURN(build_serialization());
-}
+  const LEX_STRING *user_name= row.get_column(0);
+  const LEX_STRING *privilege_type= row.get_column(1);
+  const LEX_STRING *db_name= row.get_column(2);
+  const LEX_STRING *tbl_name= row.get_column(3);
+  const LEX_STRING *col_name= row.get_column(4);
 
-/**
-  Build the serialization string.
+  LEX_STRING table_name= { C_STRING_WITH_LEN("") };
+  LEX_STRING column_name= { C_STRING_WITH_LEN("") };
 
-  This constructs the serialization string for identification
-  use in describing tablespace to the user and for creating the
-  tablespace.
+  if (tbl_name)
+    table_name= *tbl_name;
 
-  @todo take serialization_version into account
+  if (col_name)
+    column_name= *col_name;
 
-  @returns Serialization string.
-*/
-const String *TablespaceObj::build_serialization()
-{
-  DBUG_ENTER("TablespaceObj::build_serialization()");
+  m_user_name.copy(user_name->str, user_name->length, system_charset_info);
 
-  if (m_create_stmt.length())
-    DBUG_RETURN(&m_create_stmt);
+  /* Grant info. */
 
-  /*
-    Construct the CREATE TABLESPACE command from the variables.
-  */
-  m_create_stmt.length(0);
-  m_create_stmt.append("CREATE TABLESPACE ");
-  if (m_ts_name.length() > 0)
-  {
-    THD *thd= current_thd;
-    append_identifier(thd, &m_create_stmt,
-      m_ts_name.c_ptr(), m_ts_name.length());
-  }
-  m_create_stmt.append(" ADD DATAFILE '");
-  m_create_stmt.append(m_datafile);
-  if (m_comments.length())
-  {
-    m_create_stmt.append("' COMMENT = '");
-    m_create_stmt.append(m_comments);
-  }
-  m_create_stmt.append("' ENGINE=");
-  m_create_stmt.append(m_engine);
-  DBUG_RETURN(&m_create_stmt);
-}
+  String_stream s_stream(&m_grant_info);
+  s_stream << privilege_type;
 
-/**
-  Create the object.
+  if (column_name.length)
+    s_stream << "(" << &column_name << ")";
 
-  This method uses serialization string in a query and executes it.
+  s_stream << " ON " << db_name << ".";
 
-  @param[in]  thd  Thread context.
+  if (table_name.length)
+    s_stream << &table_name;
+  else
+    s_stream << "*";
 
-  @returns Error status.
-    @retval FALSE on success
-    @retval TRUE on error
-*/
-bool TablespaceObj::do_execute(THD *thd)
-{
-  DBUG_ENTER("TablespaceObj::execute()");
-  build_serialization(); // Build the CREATE command.
-  DBUG_RETURN(silent_exec(thd, &m_create_stmt));
+  /* Id. */
+
+  generate_unique_grant_id(&m_user_name, &m_id);
 }
 
-///////////////////////////////////////////////////////////////////////////
-//
-// Implementation: DbGrantObj class.
-//
 /////////////////////////////////////////////////////////////////////////////
 
-DbGrantObj::DbGrantObj(const String *grantee,
-                       const String *db_name,
-                       const String *priv_type)
-{
-  // copy strings to newly allocated memory
-  m_db_name.copy(*db_name);
-  m_grantee.copy(*grantee);
-  m_name.copy(*grantee);
-  m_priv_type.copy(*priv_type);
+Grant_obj::Grant_obj(LEX_STRING name)
+  : Abstract_obj(null_lex_str)
+{
+  m_user_name.length(0);
+  m_grant_info.length(0);
+
+  m_id.copy(name.str, name.length, system_charset_info);
 }
 
+/////////////////////////////////////////////////////////////////////////////
+
 /**
   Serialize the object.
 
   This method produces the data necessary for materializing the object
   on restore (creates object).
 
-  @param[in]  thd            Thread context.
-  @param[out] serialization  The data needed to recreate this object.
+  @param[in]  thd Thread context.
+  @param[out] out_stream  Output stream.
 
   @note this method will return an error if the db_name is either
-        mysql or information_schema as these are not objects that
-        should be recreated using this interface.
+  mysql or information_schema as these are not objects that
+  should be recreated using this interface.
 
   @returns Error status.
-    @retval FALSE on success
-    @retval TRUE on error
+    @retval FALSE on success.
+    @retval TRUE on error.
 */
-bool DbGrantObj::do_serialize(THD *thd, String *serialization)
+
+bool Grant_obj::do_serialize(THD *thd, Out_stream &out_stream)
 {
-  DBUG_ENTER("DbGrantObj::do_serialize()");
-  serialization->length(0);
-  serialization->append("GRANT ");
-  serialization->append(m_priv_type);
-  serialization->append(" ON ");
-  serialization->append(m_db_name);
-  serialization->append(".* TO ");
-  serialization->append(m_grantee);
-  DBUG_RETURN(0);
-}
+  DBUG_ENTER("Grant_obj::do_serialize");
 
-/**
-  Materialize the serialization string.
+  out_stream <<
+    &m_user_name <<
+    &m_grant_info <<
+    "SET character_set_client= binary";
 
-  This method saves serialization string into a member variable.
+  String_stream s_stream;
+  s_stream << "GRANT " << &m_grant_info << " TO " << &m_user_name;
 
-  @param[in]  serialization_version   version number of this interface
-  @param[in]  serialization           the string from serialize()
+  out_stream << s_stream;
 
-  @todo take serialization_version into account
+  DBUG_RETURN(FALSE);
+}
 
-  @returns Error status.
-    @retval FALSE on success
-    @retval TRUE on error
-*/
-bool DbGrantObj::materialize(uint serialization_version,
-                             const String *serialization)
+/////////////////////////////////////////////////////////////////////////////
+
+bool Grant_obj::do_init_from_image(In_stream *is)
 {
-  DBUG_ENTER("DbGrantObj::materialize()");
-  m_grant_stmt.copy(*serialization);
-  DBUG_RETURN(0);
-}
+  LEX_STRING user_name;
+  LEX_STRING grant_info;
 
-/**
-  Create the object.
+  if (is->next(&user_name))
+    return TRUE; /* Can not decode user name. */
 
-  This method uses serialization string in a query and executes it.
+  if (is->next(&grant_info))
+    return TRUE; /* Can not decode grant info. */
 
-  @param[in]  thd  Thread context.
+  m_user_name.copy(user_name.str, user_name.length, system_charset_info);
+  m_grant_info.copy(grant_info.str, grant_info.length, system_charset_info);
 
-  @returns Error status.
-    @retval FALSE on success
-    @retval TRUE on error
-*/
-bool DbGrantObj::do_execute(THD *thd)
-{
-  DBUG_ENTER("DbGrantObj::do_execute()");
-  DBUG_RETURN(execute_with_ctx(thd, &m_grant_stmt, true));
+  return Abstract_obj::do_init_from_image(is);
 }
 
 ///////////////////////////////////////////////////////////////////////////
-//
-// Implementation: TblGrantObj class.
-//
 /////////////////////////////////////////////////////////////////////////////
 
-TblGrantObj::TblGrantObj(const String *grantee,
-                         const String *db_name,
-                         const String *table_name,
-                         const String *priv_type)
-: DbGrantObj(grantee, db_name, priv_type)
+Obj *get_database_stub(const String *db_name)
 {
-  // copy strings to newly allocated memory
-  m_table_name.copy(*table_name);
+  return new Database_obj(db_name->lex_string());
 }
 
-/**
-  Serialize the object.
+///////////////////////////////////////////////////////////////////////////
 
-  This method produces the data necessary for materializing the object
-  on restore (creates object).
+Obj_iterator *get_databases(THD *thd)
+{
+  LEX_STRING query=
+  { C_STRING_WITH_LEN(
+                      "SELECT schema_name "
+                      "FROM INFORMATION_SCHEMA.SCHEMATA "
+                      "WHERE LCASE(schema_name) != 'mysql' AND "
+                      "LCASE(schema_name) != 'information_schema'") };
 
-  @param[in]  thd            Thread context.
-  @param[out] serialization  The data needed to recreate this object.
+  return create_row_set_iterator<Database_iterator>(thd, &query);
+}
 
-  @note this method will return an error if the db_name is either
-        mysql or information_schema as these are not objects that
-        should be recreated using this interface.
+///////////////////////////////////////////////////////////////////////////
 
-  @returns Error status.
-    @retval FALSE on success
-    @retval TRUE on error
-*/
-bool TblGrantObj::do_serialize(THD *thd, String *serialization)
+Obj_iterator *get_db_tables(THD *thd, const String *db_name)
 {
-  DBUG_ENTER("TblGrantObj::do_serialize()");
-  serialization->length(0);
-  serialization->append("GRANT ");
-  serialization->append(m_priv_type);
-  serialization->append(" ON ");
-  serialization->append(m_db_name);
-  serialization->append(".");
-  serialization->append(m_table_name);
-  serialization->append(" TO ");
-  serialization->append(m_grantee);
-  DBUG_RETURN(0);
+  String_stream s_stream;
+
+  s_stream <<
+    "SELECT '" << db_name << "', table_name "
+    "FROM INFORMATION_SCHEMA.TABLES "
+    "WHERE table_schema = '" << db_name << "' AND "
+    "table_type = 'BASE TABLE'";
+
+  return create_row_set_iterator<Db_tables_iterator>(thd, s_stream.lex_string());
 }
 
 ///////////////////////////////////////////////////////////////////////////
-//
-// Implementation: ColGrantObj class.
-//
-/////////////////////////////////////////////////////////////////////////////
 
-ColGrantObj::ColGrantObj(const String *grantee,
-                         const String *db_name,
-                         const String *table_name,
-                         const String *col_name,
-                         const String *priv_type)
-: TblGrantObj(grantee, db_name, table_name, priv_type)
+Obj_iterator *get_db_views(THD *thd, const String *db_name)
 {
-  // copy strings to newly allocated memory
-  m_col_name.copy(*col_name);
+  String_stream s_stream;
+  s_stream <<
+    "SELECT '" << db_name << "', table_name "
+    "FROM INFORMATION_SCHEMA.TABLES "
+    "WHERE table_schema = '" << db_name << "' AND table_type = 'VIEW'";
+
+  return create_row_set_iterator<Db_views_iterator>(thd, s_stream.lex_string());
 }
 
-/**
-  Serialize the object.
+///////////////////////////////////////////////////////////////////////////
 
-  This method produces the data necessary for materializing the object
-  on restore (creates object).
+Obj_iterator *get_db_triggers(THD *thd, const String *db_name)
+{
+  String_stream s_stream;
+  s_stream <<
+    "SELECT '" << db_name << "', trigger_name "
+    "FROM INFORMATION_SCHEMA.TRIGGERS "
+    "WHERE trigger_schema = '" << db_name << "'";
 
-  @param[in]  thd            Thread context.
-  @param[out] serialization  The data needed to recreate this object.
+  return create_row_set_iterator<Db_trigger_iterator>(thd, s_stream.lex_string());
+}
 
-  @note this method will return an error if the db_name is either
-        mysql or information_schema as these are not objects that
-        should be recreated using this interface.
+///////////////////////////////////////////////////////////////////////////
 
-  @returns Error status.
-    @retval FALSE on success
-    @retval TRUE on error
-*/
-bool ColGrantObj::do_serialize(THD *thd, String *serialization)
+Obj_iterator *get_db_stored_procedures(THD *thd, const String *db_name)
 {
-  DBUG_ENTER("ColGrantObj::do_serialize()");
-  serialization->length(0);
-  serialization->append("GRANT ");
-  serialization->append(m_priv_type);
-  serialization->append("(");
-  serialization->append(m_col_name);
-  serialization->append(") ON ");
-  serialization->append(m_db_name);
-  serialization->append(".");
-  serialization->append(m_table_name);
-  serialization->append(" TO ");
-  serialization->append(m_grantee);
-  DBUG_RETURN(0);
+  String_stream s_stream;
+  s_stream <<
+    "SELECT '" << db_name << "', routine_name "
+    "FROM INFORMATION_SCHEMA.ROUTINES "
+    "WHERE routine_schema = '" << db_name << "' AND "
+    "routine_type = 'PROCEDURE'";
+
+  return create_row_set_iterator<Db_stored_proc_iterator>(thd, s_stream.lex_string());
 }
 
 ///////////////////////////////////////////////////////////////////////////
 
-Obj *get_database(const String *db_name)
+Obj_iterator *get_db_stored_functions(THD *thd, const String *db_name)
 {
-  return new DatabaseObj(db_name);
-}
+  String_stream s_stream;
+  s_stream <<
+    "SELECT '" << db_name << "', routine_name "
+    "FROM INFORMATION_SCHEMA.ROUTINES "
+    "WHERE routine_schema = '" << db_name <<"' AND "
+    "routine_type = 'FUNCTION'";
 
-Obj *get_table(const String *db_name,
-               const String *table_name)
-{
-  return new TableObj(db_name, table_name, false);
+  return create_row_set_iterator<Db_stored_func_iterator>(thd, s_stream.lex_string());
 }
 
-Obj *get_view(const String *db_name,
-              const String *view_name)
-{
-  return new TableObj(db_name, view_name, true);
-}
+///////////////////////////////////////////////////////////////////////////
 
-Obj *get_trigger(const String *db_name,
-                 const String *trigger_name)
+Obj_iterator *get_db_events(THD *thd, const String *db_name)
 {
-  return new TriggerObj(db_name, trigger_name);
+#ifdef HAVE_EVENT_SCHEDULER
+  String_stream s_stream;
+  s_stream <<
+    "SELECT '" << db_name << "', event_name "
+    "FROM INFORMATION_SCHEMA.EVENTS "
+    "WHERE event_schema = '" << db_name <<"'";
+
+  return create_row_set_iterator<Db_event_iterator>(thd, s_stream.lex_string());
+#else
+  return NULL;
+#endif
 }
 
-Obj *get_stored_procedure(const String *db_name,
-                          const String *sp_name)
+///////////////////////////////////////////////////////////////////////////
+
+Obj_iterator *get_all_db_grants(THD *thd, const String *db_name)
 {
-  return new StoredProcObj(db_name, sp_name);
+  String_stream s_stream;
+  s_stream <<
+    "(SELECT grantee AS c1, "
+    "privilege_type AS c2, "
+    "table_schema AS c3, "
+    "NULL AS c4, "
+    "NULL AS c5 "
+    "FROM INFORMATION_SCHEMA.SCHEMA_PRIVILEGES "
+    "WHERE table_schema = '" << db_name << "') "
+    "UNION "
+    "(SELECT grantee, "
+    "privilege_type, "
+    "table_schema, "
+    "table_name, "
+    "NULL "
+    "FROM INFORMATION_SCHEMA.TABLE_PRIVILEGES "
+    "WHERE table_schema = '" << db_name << "') "
+    "UNION "
+    "(SELECT grantee, "
+    "privilege_type, "
+    "table_schema, "
+    "table_name, "
+    "column_name "
+    "FROM INFORMATION_SCHEMA.COLUMN_PRIVILEGES "
+    "WHERE table_schema = '" << db_name << "') "
+    "ORDER BY c1 ASC, c2 ASC, c3 ASC, c4 ASC, c5 ASC";
+
+  return create_row_set_iterator<Grant_iterator>(thd, s_stream.lex_string());
 }
 
-Obj *get_stored_function(const String *db_name,
-                         const String *sf_name)
+///////////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////
+
+Obj_iterator* get_view_base_tables(THD *thd,
+                                   const String *db_name,
+                                   const String *view_name)
 {
-  return new StoredFuncObj(db_name, sf_name);
+  return View_base_table_iterator::create(thd, db_name, view_name);
 }
 
-Obj *get_event(const String *db_name,
-               const String *event_name)
+///////////////////////////////////////////////////////////////////////////
+
+Obj_iterator* get_view_base_views(THD *thd,
+                                  const String *db_name,
+                                  const String *view_name)
 {
-#ifdef HAVE_EVENT_SCHEDULER
-  return new EventObj(db_name, event_name);
-#else
-  return NULL;
-#endif
+  return View_base_view_iterator::create(thd, db_name, view_name);
 }
 
-
+///////////////////////////////////////////////////////////////////////////
 ///////////////////////////////////////////////////////////////////////////
 
-Obj *materialize_database(const String *db_name,
-                          uint serialization_version,
-                          const String *serialization)
+Obj *get_database(const String *db_name,
+                  uint image_version,
+                  const String *image)
 {
-  Obj *obj= new DatabaseObj(db_name);
-  obj->materialize(serialization_version, serialization);
+  Database_obj *obj= new Database_obj(db_name->lex_string());
+  obj->init_from_image(image_version, image);
 
   return obj;
 }
 
-Obj *materialize_table(const String *db_name,
-                       const String *table_name,
-                       uint serialization_version,
-                       const String *serialization)
-{
-  Obj *obj= new TableObj(db_name, table_name, false);
-  obj->materialize(serialization_version, serialization);
+///////////////////////////////////////////////////////////////////////////
+
+Obj *get_table(const String *db_name,
+               const String *table_name,
+               uint image_version,
+               const String *image)
+{
+  Table_obj *obj= new Table_obj(db_name->lex_string(),
+                                table_name->lex_string());
+  obj->init_from_image(image_version, image);
 
   return obj;
 }
 
-Obj *materialize_view(const String *db_name,
-                      const String *view_name,
-                      uint serialization_version,
-                      const String *serialization)
+///////////////////////////////////////////////////////////////////////////
+
+Obj *get_view(const String *db_name,
+              const String *view_name,
+              uint image_version,
+              const String *image)
 {
-  Obj *obj= new TableObj(db_name, view_name, true);
-  obj->materialize(serialization_version, serialization);
+  View_obj *obj= new View_obj(db_name->lex_string(), view_name->lex_string());
+  obj->init_from_image(image_version, image);
 
   return obj;
 }
 
-Obj *materialize_trigger(const String *db_name,
-                         const String *trigger_name,
-                         uint serialization_version,
-                         const String *serialization)
-{
-  Obj *obj= new TriggerObj(db_name, trigger_name);
-  obj->materialize(serialization_version, serialization);
+///////////////////////////////////////////////////////////////////////////
+
+Obj *get_trigger(const String *db_name,
+                 const String *trigger_name,
+                 uint image_version,
+                 const String *image)
+{
+  Trigger_obj *obj= new Trigger_obj(db_name->lex_string(),
+                                    trigger_name->lex_string());
+  if (obj->init_from_image(image_version, image))
+  {
+    delete obj;
+    obj= 0;
+  }
 
   return obj;
 }
 
-Obj *materialize_stored_procedure(const String *db_name,
-                                  const String *stored_proc_name,
-                                  uint serialization_version,
-                                  const String *serialization)
-{
-  Obj *obj= new StoredProcObj(db_name, stored_proc_name);
-  obj->materialize(serialization_version, serialization);
+///////////////////////////////////////////////////////////////////////////
+
+Obj *get_stored_procedure(const String *db_name,
+                          const String *sp_name,
+                          uint image_version,
+                          const String *image)
+{
+  Stored_proc_obj *obj= new Stored_proc_obj(db_name->lex_string(),
+                                            sp_name->lex_string());
+  if (obj->init_from_image(image_version, image))
+  {
+    delete obj;
+    obj= 0;
+  }
 
   return obj;
 }
 
-Obj *materialize_stored_function(const String *db_name,
-                                 const String *stored_func_name,
-                                 uint serialization_version,
-                                 const String *serialization)
-{
-  Obj *obj= new StoredFuncObj(db_name, stored_func_name);
-  obj->materialize(serialization_version, serialization);
+///////////////////////////////////////////////////////////////////////////
+
+Obj *get_stored_function(const String *db_name,
+                         const String *sf_name,
+                         uint image_version,
+                         const String *image)
+{
+  Stored_func_obj *obj= new Stored_func_obj(db_name->lex_string(),
+                                            sf_name->lex_string());
+  if (obj->init_from_image(image_version, image))
+  {
+    delete obj;
+    obj= 0;
+  }
 
   return obj;
 }
 
+///////////////////////////////////////////////////////////////////////////
+
 #ifdef HAVE_EVENT_SCHEDULER
-Obj *materialize_event(const String *db_name,
-                       const String *event_name,
-                       uint serialization_version,
-                       const String *serialization)
-{
-  Obj *obj= new EventObj(db_name, event_name);
-  obj->materialize(serialization_version, serialization);
+
+Obj *get_event(const String *db_name,
+               const String *event_name,
+               uint image_version,
+               const String *image)
+{
+  Event_obj *obj= new Event_obj(db_name->lex_string(),
+                                event_name->lex_string());
+  if (obj->init_from_image(image_version, image))
+  {
+    delete obj;
+    obj= 0;
+  }
 
   return obj;
 }
+
 #endif
 
-Obj *materialize_tablespace(const String *ts_name,
-                            uint serialization_version,
-                            const String *serialization)
+///////////////////////////////////////////////////////////////////////////
+
+Obj *get_tablespace(const String *ts_name,
+                    uint image_version,
+                    const String *image)
 {
-  Obj *obj= new TablespaceObj(ts_name);
-  obj->materialize(serialization_version, serialization);
+  Tablespace_obj *obj= new Tablespace_obj(ts_name->lex_string());
+  if (obj->init_from_image(image_version, image))
+  {
+    delete obj;
+    obj= 0;
+  }
 
   return obj;
 }
 
-Obj *get_db_grant(const String *grantee,
-                  const String *db_name)
-{
-  String priv_type;
-  priv_type.length(0);
-
-  return new DbGrantObj(grantee, db_name, &priv_type);
-}
+///////////////////////////////////////////////////////////////////////////
 
-Obj *materialize_db_grant(const String *db_name,
-                          const String *grantee,
-                          uint serialization_version,
-                          const String *serialization)
+Obj *get_db_grant(const String *db_name,
+                  const String *name,
+                  uint image_version,
+                  const String *image)
 {
-  /*
-    Here we create a grant for the purposes of applying the
-    grants. We use DbGrantObj for all types of grants because
-    we only have the GRANT statement in the serialization
-    string and therefore do not that the 'parts' to create
-    the specific types.
-  */
-  Obj *obj= get_db_grant(grantee, db_name);
-  obj->materialize(serialization_version, serialization);
+  Grant_obj *obj= new Grant_obj(name->lex_string());
+  if (obj->init_from_image(image_version, image))
+  {
+    delete obj;
+    obj= 0;
+  }
 
   return obj;
 }
 
 ///////////////////////////////////////////////////////////////////////////
+///////////////////////////////////////////////////////////////////////////
 
 bool is_internal_db_name(const String *db_name)
 {
@@ -3319,370 +2637,190 @@ bool is_internal_db_name(const String *d
 
 ///////////////////////////////////////////////////////////////////////////
 
-bool check_db_existence(const String *db_name)
+bool check_db_existence(THD *thd, const String *db_name)
 {
-  return check_db_dir_existence(((String *) db_name)->c_ptr_safe());
+  String_stream s_stream;
+  int rc;
+
+  s_stream << "SHOW CREATE DATABASE `" << db_name << "`";
+
+  Ed_result ed_result;
+  rc= run_service_interface_sql(thd, s_stream.lex_string(), &ed_result);
+
+  /* We're not interested in warnings/errors here. */
+
+  return test(rc);
 }
 
-/*
-  Splits grantee clause into user and host portions. Needed for checking
-  to see if user exists on system.
-*/
-int split_user_host(String *grantee, String *user, String *host)
+///////////////////////////////////////////////////////////////////////////
+
+bool check_user_existence(THD *thd, const Obj *obj)
 {
-  int len= 0;
-  int tics= 0;
-  char *ptr= 0;
+#ifdef EMBEDDED_LIBRARY
+  return TRUE;
+#else
+  Grant_obj *grant_obj= (Grant_obj *) obj;
+  Ed_result ed_result;
+  String_stream s_stream;
 
-  /*
-    Since passwords are single byte characters and usernames can be multibyte
-    characters and the 0x40 = 64 = @ can occur in the username, we must search
-    for the first @ from the right.
-  */
-  len= grantee->length();
-  len--;
-  ptr= grantee->c_ptr() + len;
-  while ((len > 0) && (*ptr != '@'))
-  {
-    len--;
-    ptr= grantee->c_ptr() + len;
-  }
+  s_stream <<
+    "SELECT 1 "
+    "FROM INFORMATION_SCHEMA.USER_PRIVILEGES "
+    "WHERE grantee = \"" << grant_obj->get_user_name() << "\"";
 
-  if (ptr == 0)
-    return -1;
-  len= ptr - grantee->c_ptr();
-  user->length(0);
-  char *cptr= grantee->c_ptr();
 
-  /*
-    String ' from strings.
-  */
-  if (strncmp(cptr, "'", 1) == 0)
+  if (run_service_interface_sql(thd, s_stream.lex_string(), &ed_result) ||
+      ed_result.get_warnings().elements > 0)
   {
-    cptr++;
-    len--;
-    tics++;
-  }
-  user->append(cptr, len - tics);
-  len= grantee->length() - len - 1 - tics;
-  host->length(0);
-
-  /*
-    String ' from strings.
-  */
-  cptr= ptr + 1;
-  tics= 0;
-  if (strncmp(cptr, "'", 1) == 0)
-  {
-    cptr++;
-    len--;
-  }
-  if (strncmp(cptr+len-1, "'", 1) == 0)
-    tics++;
-  host->append(cptr, len - tics);
-  return 0;
-}
-
-bool check_user_existence(THD *thd, const String *grantee)
-{
-  String user;
-  String host;
-  bool user_exists= FALSE;
-
-  user.length(0);
-  host.length(0);
-  if (grantee)
-  {
-#ifndef EMBEDDED_LIBRARY
-    split_user_host((String *)grantee, &user, &host);
-    if (!user.ptr())
-      user.append("''");
-    user_exists= is_acl_user(host.ptr(), user.ptr());
-#else
-    user_exists= TRUE;
-#endif
+    /* Should be no warnings. */
+    return FALSE;
   }
-  return user_exists;
-}
-
-/**
-  Locate the row in the information_schema view for this tablespace.
 
-  This method returns a row from a tablespace information_schema view
-  that matches the tablespace name passed.
+  Ed_result_set *ed_result_set= ed_result.get_cur_result_set();
 
-  @param[in]     thd           Thread context
-  @param[in]     is_table_idx  The information schema to search
-  @param[in]     ts_name       The name of the tablespace to find
-  @param[in]     ts_engine     Engine of the tablespace to find
-  @param[out]    datafile      The datafile for the tablespace
-  @param[out]    comments      The comments for the tablespace
-
-  @retval FALSE if tablespace exists and no errors
-  @retval TRUE if tablespace does not exist or errors
-*/
-static bool find_tablespace_schema_row(THD *thd,
-                                       enum_schema_tables is_table_idx,
-                                       const String *ts_name,
-                                       const String *ts_engine,
-                                       String *datafile,
-                                       String *comments)
-{
-  TABLE *is_table;
-  handler *ha;
-  my_bitmap_map *orig_col;
-  LEX_STRING lex_ts_name;
-  String found_ts_name, found_ts_engine;
-  bool retval= TRUE;
-  String data;
-  List<LEX_STRING> ts_list;
-  DBUG_ENTER("obs::find_tablespace_schema_row()");
+  if (!ed_result_set)
+    return FALSE;
 
-  /*
-    First, open the IS table.
-  */
-  lex_ts_name.str= (char *)ts_name->ptr();
-  lex_ts_name.length= ts_name->length();
-  ts_list.push_back(&lex_ts_name);
+  return ed_result_set->data()->elements > 0;
+#endif
+}
 
-  if (InformationSchemaIterator::prepare_is_table(
-      thd, &is_table, &ha, &orig_col, is_table_idx, ts_list))
-    DBUG_RETURN(TRUE);
+///////////////////////////////////////////////////////////////////////////
 
-  /* Locate the row in the schema table and retrive the data. */
-  switch (is_table_idx) {
-  case SCH_TABLESPACES:
-    while (!ha->rnd_next(is_table->record[0]))
-    {
-      is_table->field[IS_TABLESPACES_TABLESPACE_NAME]->val_str(&found_ts_name);
-      is_table->field[IS_TABLESPACES_ENGINE]->val_str(&found_ts_engine);
-      if (found_ts_name.length() && found_ts_engine.length() &&
-          !my_strnncoll(system_charset_info, (const uchar*) found_ts_name.ptr(),
-                        found_ts_name.length(), (const uchar*) ts_name->ptr(),
-                        ts_name->length()) &&
-          !my_strnncoll(system_charset_info,
-                        (const uchar*) found_ts_engine.ptr(),
-                        found_ts_engine.length(),
-                        (const uchar*) ts_engine->ptr(), ts_engine->length()))
-      {
-        retval= FALSE;
-        is_table->field[IS_TABLESPACES_TABLESPACE_COMMENT]->val_str(&data);
-        comments->copy(data);
-        DBUG_PRINT("find_tablespace_schema_row", (" Found tablespace %s",
-                   found_ts_name.ptr()));
-        break;
-      }
-      found_ts_name.length(0);
-      found_ts_engine.length(0);
-    }
-    break;
-  case SCH_FILES:
-    while (!ha->rnd_next(is_table->record[0]))
-    {
-      is_table->field[IS_FILES_TABLESPACE_NAME]->val_str(&found_ts_name);
-      is_table->field[IS_FILES_ENGINE]->val_str(&found_ts_engine);
-      if (found_ts_name.length() && found_ts_engine.length() &&
-          !my_strnncoll(system_charset_info, (const uchar*) found_ts_name.ptr(),
-                        found_ts_name.length(), (const uchar*) ts_name->ptr(),
-                        ts_name->length()) &&
-          !my_strnncoll(system_charset_info,
-                        (const uchar*) found_ts_engine.ptr(),
-                        found_ts_engine.length(),
-                        (const uchar*) ts_engine->ptr(), ts_engine->length()))
-      {
-        retval= FALSE;
-        is_table->field[IS_FILES_FILE_NAME]->val_str(&data);
-        datafile->copy(data);
-        DBUG_PRINT("find_tablespace_schema_row", (" Found tablespace %s",
-                   found_ts_name.ptr()));
-        break;
-      }
-      found_ts_name.length(0);
-      found_ts_engine.length(0);
-    }
-    break;
-  default:
-    DBUG_ASSERT(0);
-    break;
-  }
+const String *grant_get_user_name(const Obj *obj)
+{
+  return ((Grant_obj *) obj)->get_user_name();
+}
 
-  /*
-    Cleanup
-  */
-  ha->ha_rnd_end();
+///////////////////////////////////////////////////////////////////////////
 
-  dbug_tmp_restore_column_map(is_table->read_set, orig_col);
-  free_tmp_table(thd, is_table);
-  DBUG_RETURN(retval);
+const String *grant_get_grant_info(const Obj *obj)
+{
+  return ((Grant_obj *) obj)->get_grant_info();
 }
 
-/**
-  Build a valid tablespace from the information_schema views.
+///////////////////////////////////////////////////////////////////////////
 
-  This method builds a @c TablespaceObj object if the tablespace
-  exists on the server.
+Obj *find_tablespace(THD *thd, const String *ts_name)
+{
+  Ed_result ed_result;
+  String_stream s_stream;
 
-  @param[in]     thd           Thread context.
-  @param[out]    TablespaceObj A pointer to a new tablespace object
-  @param[in]     ts_name       The name of the tablespace to find
-  @param[in]     ts_engine     Engine of the tablespace to find
+  s_stream <<
+    "SELECT t1.tablespace_comment, t2.file_name, t1.engine "
+    "FROM INFORMATION_SCHEMA.TABLESPACES AS t1, "
+    "INFORMATION_SCHEMA.FILES AS t2 "
+    "WHERE t1.tablespace_name = t2.tablespace_name AND "
+    "t1.tablespace_name = '" << ts_name << "'";
 
-  @note Caller is responsible for destroying the tablespace object.
 
-  @retval FALSE if tablespace exists and no errors
-  @retval TRUE if tablespace does not exist or errors
-*/
-static bool get_tablespace_from_schema(THD *thd,
-                                       TablespaceObj **ts,
-                                       const String *ts_name,
-                                       const String *ts_engine)
-{
-  String datafile;
-  String comments;
-  DBUG_ENTER("obs::get_tablespace_from_schema()");
+  if (run_service_interface_sql(thd, s_stream.lex_string(), &ed_result) ||
+      ed_result.get_warnings().elements > 0)
+  {
+    /* Should be no warnings. */
+    return NULL;
+  }
 
-  /*
-    Locate the row in TABLESPACES and get the comments.
-  */
-  if (find_tablespace_schema_row(thd, SCH_TABLESPACES,
-      ts_name, ts_engine, &datafile, &comments))
-    DBUG_RETURN(TRUE);
+  if (!ed_result.elements)
+    return NULL;
 
-  /*
-    Locate the row in FILES and get the datafile.
-  */
-  if (find_tablespace_schema_row(thd, SCH_FILES,
-      ts_name, ts_engine, &datafile, &comments))
-    DBUG_RETURN(TRUE);
+  Ed_result_set *ed_result_set= ed_result.get_cur_result_set();
 
-  /*
-    The datafile parameter is required.
-  */
-  if (datafile.length() == 0)
-    DBUG_RETURN(TRUE);
+  /* The result must contain only one result-set. */
+  DBUG_ASSERT(ed_result_set->data()->elements == 1);
 
-  DBUG_PRINT("get_tablespace_from_schema", (" Found tablespace %s %s",
-    ts_name->ptr(), datafile.ptr()));
+  Ed_row *row= ed_result_set->get_cur_row();
 
-  TablespaceObj *ts_local= new TablespaceObj(ts_name);
-  *ts= ts_local;
-  ts_local->set_datafile(&datafile);
-  ts_local->set_comments(&comments);
-  ts_local->set_engine(ts_engine);
+  /* There must be 3 columns. */
+  DBUG_ASSERT(row->get_metadata()->get_num_columns() == 3);
 
-  DBUG_RETURN(FALSE);
+  const LEX_STRING *comment= row->get_column(0);
+  const LEX_STRING *data_file_name= row->get_column(1);
+  const LEX_STRING *engine= row->get_column(2);
+
+  return new Tablespace_obj(ts_name->lex_string(),
+                            *comment, *data_file_name, *engine);
 }
 
+///////////////////////////////////////////////////////////////////////////
+
 /**
   Retrieve the tablespace for a table if it exists
 
-  This method returns a @c TablespaceObj object if the table has a tablespace.
+  This method returns a @c Tablespace_obj object if the table has a tablespace.
 
-  @param[in]  thd       Thread context.
-  @param[in]  db_name   The database name for the table.
-  @param[in]  tbl_name  The table name.
+  @param[in]  thd         Thread context.
+  @param[in]  db_name     The database name for the table.
+  @param[in]  table_name  The table name.
 
   @note Caller is responsible for destroying the object.
 
-  @retval Tablespace object if table uses a tablespace
-  @retval NULL if table does not use a tablespace
+  @return Tablespace object.
+    @retval Tablespace object if table uses a tablespace.
+    @retval NULL if table does not use a tablespace.
 */
-Obj *get_tablespace_for_table(THD *thd,
-                              const String *db_name,
-                              const String *tbl_name)
+
+Obj *find_tablespace_for_table(THD *thd,
+                               const String *db_name,
+                               const String *table_name)
 {
-  TablespaceObj *ts= NULL;
-  char path[FN_REFLEN];
-  String ts_name, ts_engine;
-  const char *ts_name_str= NULL;
-  DBUG_ENTER("obs::get_tablespace_for_table()");
-  DBUG_PRINT("obs::get_tablespace_for_table", ("name: %s.%s",
-             db_name->ptr(), tbl_name->ptr()));
-
-  const char *db= db_name->ptr();
-  const char *name= tbl_name->ptr();
-
-  build_table_filename(path, sizeof(path), db, name, "", 0);
-  ts_name.length(0);
-  ts_engine.length(0);
-
-  TABLE *table= open_temporary_table(thd, path, db, name,
-                    FALSE /* don't link to thd->temporary_tables */,
-                    OTM_OPEN);
+  Ed_result ed_result;
+  String_stream s_stream;
+
+  s_stream <<
+    "SELECT t1.tablespace_name, t1.engine, t1.tablespace_comment, t2.file_name "
+    "FROM INFORMATION_SCHEMA.TABLESPACES AS t1, "
+    "INFORMATION_SCHEMA.FILES AS t2, "
+    "INFORMATION_SCHEMA.TABLES AS t3 "
+    "WHERE t1.tablespace_name = t2.tablespace_name AND "
+    "t2.tablespace_name = t3.tablespace_name AND "
+    "t3.table_schema = '" << db_name << "' AND "
+    "t3.table_name = '" << table_name << "'";
 
-  if (table)
+
+  if (run_service_interface_sql(thd, s_stream.lex_string(), &ed_result) ||
+      ed_result.get_warnings().elements > 0)
   {
-    if ((ts_name_str= table->file->get_tablespace_name()))
-    {
-      ts_name.append(ts_name_str);
-      ts_name.set_charset(system_charset_info);
-      ts_engine.append(table->file->engine_name()->str);
-      ts_engine.set_charset(system_charset_info);
-    }
-    intern_close_table(table);
-    my_free(table, MYF(0));
+    /* Should be no warnings. */
+    return NULL;
   }
-  else
-    goto end;
 
-  /*
-    Now open the information_schema table and get the tablespace information.
-  */
-  if (ts_name_str)
-    get_tablespace_from_schema(thd, &ts, &ts_name, &ts_engine);
-end:
-  DBUG_RETURN(ts);
-}
+  if (!ed_result.elements)
+    return NULL;
 
-/**
-  Determine if tablespace exists.
+  Ed_result_set *ed_result_set= ed_result.get_cur_result_set();
 
-  This method determines if a materialized tablespace exists on the
-  system. This compares the name and all saved attributes of the
-  tablespace. A FALSE return would mean either the tablespace does
-  not exist or the tablespace attributes are different.
+  if (!ed_result_set->data()->elements)
+    return NULL;
 
-  @param[in]  Obj  The TablspaceObj pointer to compare.
+  /* The result must contain only one result-set. */
+  DBUG_ASSERT(ed_result_set->data()->elements == 1);
 
-  @retval TRUE if it exists
-  @retval FALSE if it does not exist
-*/
-bool tablespace_exists(THD *thd,
-                       Obj *ts)
-{
-  TablespaceObj *other_ts= NULL, *this_ts= static_cast<TablespaceObj*>(ts);
-  bool retval= FALSE;
-  DBUG_ENTER("obs::tablespace_exists()");
-  get_tablespace_from_schema(thd, &other_ts, this_ts->get_name(),
-                             this_ts->get_engine());
-  if (!other_ts)
-    DBUG_RETURN(retval);
-  retval= (my_strcasecmp(system_charset_info,
-           other_ts->build_serialization()->ptr(),
-           ((TablespaceObj *)ts)->build_serialization()->ptr()) == 0);
-  delete other_ts;
-  DBUG_RETURN(retval);
-}
+  Ed_row *row= ed_result_set->get_cur_row();
 
-/**
-  Is there a tablespace with the given name?
+  /* There must be 4 columns. */
+  DBUG_ASSERT(row->get_metadata()->get_num_columns() == 4);
 
-  This method determines if the tablespace referenced by name exists on the
-  system. Returns a TablespaceObj if it exists or NULL if it doesn't.
+  const LEX_STRING *ts_name= row->get_column(0);
+  const LEX_STRING *engine= row->get_column(1);
+  const LEX_STRING *comment= row->get_column(2);
+  const LEX_STRING *data_file_name= row->get_column(3);
 
-  @param[in]  Obj  The TablspaceObj pointer to compare.
+  return new Tablespace_obj(*ts_name, *comment, *data_file_name, *engine);
+}
 
-  @note Caller is responsible for destroying the tablespace object.
+///////////////////////////////////////////////////////////////////////////
 
-  @returns the tablespace if found or NULL if not found
-*/
-Obj *is_tablespace(THD *thd, Obj *ts)
+bool compare_tablespace_attributes(Obj *ts1, Obj *ts2)
 {
-  TablespaceObj *other_ts= NULL, *this_ts= static_cast<TablespaceObj*>(ts);
-  DBUG_ENTER("obs::is_tablespace()");
-  get_tablespace_from_schema(thd, &other_ts, this_ts->get_name(),
-                             this_ts->get_engine());
-  DBUG_RETURN(other_ts);
+  DBUG_ENTER("obs::compare_tablespace_attributes");
+
+  Tablespace_obj *o1= (Tablespace_obj *) ts1;
+  Tablespace_obj *o2= (Tablespace_obj *) ts2;
+
+  DBUG_RETURN(my_strcasecmp(system_charset_info,
+                            o1->get_description()->ptr(),
+                            o2->get_description()->ptr()) == 0);
 }
 
 ///////////////////////////////////////////////////////////////////////////
@@ -3698,65 +2836,70 @@ Obj *is_tablespace(THD *thd, Obj *ts)
 */
 
 /**
-   Turn on the ddl blocker
+  Turn on the ddl blocker
 
-   This method is used to start the ddl blocker blocking DDL commands.
+  This method is used to start the ddl blocker blocking DDL commands.
 
-   @param[in] thd  current thread
+  @param[in] thd  current thread
 
-   @retval FALSE on success.
-   @retval TRUE on error.
-  */
+  @return Error status.
+    @retval FALSE on success.
+    @retval TRUE on error.
+*/
 bool ddl_blocker_enable(THD *thd)
 {
-  DBUG_ENTER("ddl_blocker_enable()");
+  DBUG_ENTER("obs::ddl_blocker_enable");
   if (!DDL_blocker->block_DDL(thd))
     DBUG_RETURN(TRUE);
   DBUG_RETURN(FALSE);
 }
 
+
 /**
-   Turn off the ddl blocker
+  Turn off the ddl blocker
 
-   This method is used to stop the ddl blocker from blocking DDL commands.
-  */
+  This method is used to stop the ddl blocker from blocking DDL commands.
+*/
 void ddl_blocker_disable()
 {
-  DBUG_ENTER("ddl_blocker_disable()");
+  DBUG_ENTER("obs::ddl_blocker_disable");
   DDL_blocker->unblock_DDL();
   DBUG_VOID_RETURN;
 }
 
+
 /**
-   Turn on the ddl blocker exception
+  Turn on the ddl blocker exception
 
-   This method is used to allow the exception allowing a restore operation to
-   perform DDL operations while the ddl blocker blocking DDL commands.
+  This method is used to allow the exception allowing a restore operation to
+  perform DDL operations while the ddl blocker blocking DDL commands.
 
-   @param[in] thd  current thread
-  */
+  @param[in] thd  current thread
+*/
 void ddl_blocker_exception_on(THD *thd)
 {
-  DBUG_ENTER("ddl_blocker_exception_on()");
+  DBUG_ENTER("obs::ddl_blocker_exception_on");
   thd->DDL_exception= TRUE;
   DBUG_VOID_RETURN;
 }
 
+
 /**
-   Turn off the ddl blocker exception
+  Turn off the ddl blocker exception
 
-   This method is used to suspend the exception allowing a restore operation to
-   perform DDL operations while the ddl blocker blocking DDL commands.
+  This method is used to suspend the exception allowing a restore operation to
+  perform DDL operations while the ddl blocker blocking DDL commands.
 
-   @param[in] thd  current thread
-  */
+  @param[in] thd  current thread
+*/
 void ddl_blocker_exception_off(THD *thd)
 {
-  DBUG_ENTER("ddl_blocker_exception_off()");
+  DBUG_ENTER("obs::ddl_blocker_exception_off");
   thd->DDL_exception= FALSE;
   DBUG_VOID_RETURN;
 }
 
+
 /**
   Build a table list from a list of tables as class Obj.
 
@@ -3774,7 +2917,7 @@ TABLE_LIST *Name_locker::build_table_lis
 {
   TABLE_LIST *tl= NULL;
   Obj *tbl= NULL;
-  DBUG_ENTER("Name_locker::build_table_list()");
+  DBUG_ENTER("Name_locker::build_table_list");
 
   List_iterator<Obj> it(*tables);
   while ((tbl= it++))
@@ -3798,15 +2941,16 @@ TABLE_LIST *Name_locker::build_table_lis
   DBUG_RETURN(tl);
 }
 
-void Name_locker::free_table_list(TABLE_LIST *tl)
+
+void Name_locker::free_table_list(TABLE_LIST *table_list)
 {
-  TABLE_LIST *ptr= tl;
+  TABLE_LIST *ptr= table_list;
 
   while (ptr)
   {
-    tl= tl->next_global;
+    table_list= table_list->next_global;
     my_free(ptr, MYF(0));
-    ptr= tl;
+    ptr= table_list;
   }
 }
 
@@ -3825,7 +2969,7 @@ int Name_locker::get_name_locks(List<Obj
 {
   TABLE_LIST *ltable= 0;
   int ret= 0;
-  DBUG_ENTER("Name_locker::get_name_locks()");
+  DBUG_ENTER("Name_locker::get_name_locks");
   /*
     Convert List<Obj> to TABLE_LIST *
   */
@@ -3852,7 +2996,7 @@ int Name_locker::get_name_locks(List<Obj
 */
 int Name_locker::release_name_locks()
 {
-  DBUG_ENTER("Name_locker::release_name_locks()");
+  DBUG_ENTER("Name_locker::release_name_locks");
   if (m_table_list)
   {
     pthread_mutex_unlock(&LOCK_open);
@@ -3861,6 +3005,6 @@ int Name_locker::release_name_locks()
   DBUG_RETURN(0);
 }
 
-} // obs namespace
-
 ///////////////////////////////////////////////////////////////////////////
+
+} // obs namespace

=== modified file 'sql/si_objects.h'
--- a/sql/si_objects.h	2008-10-27 13:06:21 +0000
+++ b/sql/si_objects.h	2008-11-27 16:45:16 +0000
@@ -1,222 +1,124 @@
 #ifndef SI_OBJECTS_H_
 #define SI_OBJECTS_H_
 
-/**
-   @file
+/* Copyright (C) 2008 MySQL AB
 
-   This file defines the API for the following object services:
-     - serialize database objects into a string;
-     - materialize (deserialize) object from a string;
-     - enumerating objects;
-     - finding dependencies for objects;
-     - executor for SQL statements;
-     - wrappers for controlling the DDL Blocker;
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; version 2 of the License.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307 USA */
+
+/**
+  @file
+
+  An object in this file refers to SQL language database objects
+  such as tables, views, stored programs, events or users,
+  databases and tablespaces.
+  This file defines an API for the following object services:
+    - serialize object definition (metadata) into a string,
+    - materialize (de-serialize) object from a string,
+    - enumerate all objects of a database,
+    - find dependencies between objects, such as underlying tables of
+    views, tablespace of a table.
+
+  Additionally, the interface provides two helper services
+  for backup:
+    - execute an arbitrary SQL statement
+    - lock and unlock all metadata, so called "DDL blocker"
 */
 
 namespace obs {
 
+///////////////////////////////////////////////////////////////////////////
+
 /**
-  Obj defines the basic set of operations for each database object.
+  @class Obj
+
+  This interface defines the basic set of operations for each database object.
 */
 
 class Obj
 {
 public:
-
-  bool serialize(THD *thd, String *serialization);
-
   /**
     Return the name of the object.
 
     @return object name.
   */
-  virtual const String *get_name() = 0;
+  virtual const String *get_name() const = 0;
 
   /**
     Return the database name of the object.
 
     @note this is a subject to remove.
   */
-  virtual const String *get_db_name() = 0;
-
-  bool execute(THD *thd);
+  virtual const String *get_db_name() const = 0;
 
 public:
-  virtual ~Obj()
-  { }
-
-private:
-
   /**
-    Read the object state from a given buffer and restores object state to
-    the point, where it can be executed.
+    Serialize an object to an image. The serialization image is opaque
+    object for the client. The client should supply a valid string
+    instance, which will contain serialization image after return.
 
-    @param[in] serialization_version The version of the serialization format.
-    @param[in] serialization         Buffer contained serialized object.
+    @param[in]  thd     Thread context.
+    @param[out] image   Serialization image.
 
-    @return error status.
-      @retval FALSE on success.
-      @retval TRUE on error.
+    @return Error status.
   */
-  virtual bool materialize(uint serialization_version,
-                           const String *serialization) = 0;
+  virtual bool serialize(THD *thd, String *image) = 0;
 
-  /// Primitive implementing @c serialize() method.
-  virtual bool do_serialize(THD *thd, String *serialization) = 0;
+  /**
+    Create an object persistently in the database.
 
-  /// Primitive implementing @c execute() method.
-  virtual bool do_execute(THD *thd) = 0;
+    @param[in]  thd     Thread context.
+
+    @return Error status.
+  */
+  virtual bool create(THD *thd) = 0;
 
   /**
-    Drop the object.
+    Drop an object in the database.
 
-    @param[in] thd              Server thread context.
+    @param[in]  thd     Thread context.
 
-    @return error status.
-      @retval FALSE on success.
-      @retval TRUE on error.
+    @return Error status.
   */
   virtual bool drop(THD *thd) = 0;
 
-private:
-  friend Obj *materialize_database(const String *,
-                                   uint,
-                                   const String *);
-
-  friend Obj *materialize_table(const String *,
-                                const String *,
-                                uint,
-                                const String *);
-
-  friend Obj *materialize_view(const String *,
-                               const String *,
-                               uint,
-                               const String *);
-
-  friend Obj *materialize_trigger(const String *,
-                                  const String *,
-                                  uint,
-                                  const String *);
-
-  friend Obj *materialize_stored_procedure(const String *,
-                                           const String *,
-                                           uint,
-                                           const String *);
-
-  friend Obj *materialize_stored_function(const String *,
-                                          const String *,
-                                          uint,
-                                          const String *);
-
-  friend Obj *materialize_event(const String *,
-                                const String *,
-                                uint,
-                                const String *);
-
-  friend Obj *materialize_tablespace(const String *,
-                                     uint,
-                                     const String *);
-
-  friend Obj *materialize_db_grant(const String *,
-                                   const String *,
-                                   uint,
-                                   const String *);
-
+public:
+  virtual ~Obj()
+  { }
 };
 
-
-/**
-  Create the object in the database.
-
-  @param[in] thd              Server thread context.
-
-  @return error status.
-    @retval FALSE on success.
-    @retval TRUE on error.
-
-  @note The real work is done inside @c do_execute() primitive which should be
-  defied in derived classes. This method prepares appropriate context and calls
-  the primitive.
-*/
-inline
-bool Obj::execute(THD *thd)
-{
-  ulong saved_sql_mode= thd->variables.sql_mode;
-  thd->variables.sql_mode= 0;
-
-  set_var_collation_client saved_charset_settings(
-                             thd->variables.character_set_client,
-                             thd->variables.character_set_results,
-                             thd->variables.collation_connection);
-
-  set_var_collation_client new_charset_settings(::system_charset_info,
-                                                ::system_charset_info,
-                                                ::system_charset_info);
-  new_charset_settings.update(thd);
-
-  bool ret= do_execute(thd);
-
-  saved_charset_settings.update(thd);
-  thd->variables.sql_mode= saved_sql_mode;
-
-  return ret;
-}
-
-/**
-  Serialize object state into a buffer. The buffer actually should be a
-  binary buffer. String class is used here just because we don't have
-  convenient primitive for binary buffers.
-
-  Serialization format is opaque to the client, i.e. the client should
-  not make any assumptions about the format or the content of the
-  returned buffer.
-
-  Serialization format can be changed in the future versions. However,
-  the server must be able to materialize objects coded in any previous
-  formats.
-
-  @param[in] thd           Server thread context.
-  @param[in] serialization Buffer to serialize the object
-
-  @return error status.
-    @retval FALSE on success.
-    @retval TRUE on error.
-
-  @note The real work is done inside @c do_serialize() primitive which should be
-  defied in derived classes. This method prepares appropriate context and calls
-  the primitive.
-*/
-inline
-bool Obj::serialize(THD *thd, String *serialization)
-{
-  ulong saved_sql_mode= thd->variables.sql_mode;
-  thd->variables.sql_mode= 0;
-
-  bool ret= do_serialize(thd, serialization);
-
-  thd->variables.sql_mode= saved_sql_mode;
-
-  return ret;
-}
-
 ///////////////////////////////////////////////////////////////////////////
 
 /**
-  Obj_iterator is a basic interface to enumerate the objects.
+  @class Obj_iterator
+
+  This is a basic interface to enumerate objects.
 */
 
 class Obj_iterator
 {
 public:
-
   Obj_iterator()
   { }
 
   /**
     This operation returns a pointer to the next object in an enumeration.
     It returns NULL if there is no more objects.
+    Results of an attempt to continue iteration when there is no
+    more objects are undefined.
 
-    The client is responsible to destroy the returned object.
+    The client is responsible for destruction of the returned object.
 
     @return a pointer to the object
       @retval NULL if there is no more objects in an enumeration.
@@ -226,254 +128,115 @@ public:
 public:
   virtual ~Obj_iterator()
   { }
-
-};
-
-/**
-  GrantObjIternator is an encapsulation of the three iterators for each level
-  of grant supported: database-, table- and routine-, and column-level.
-*/
-class GrantObjIterator : public Obj_iterator
-{
-public:
-  GrantObjIterator(THD *thd, const String *db_name);
-
-  ~GrantObjIterator()
-  {
-    delete db_grants;
-    delete tbl_grants;
-    delete col_grants;
-  }
-
-  /**
-    This operation returns a pointer to the next object in an enumeration.
-    It returns NULL if there is no more objects.
-
-    The client is responsible to destroy the returned object.
-
-    @return a pointer to the object
-      @retval NULL if there is no more objects in an enumeration.
-  */
-  Obj *next();
-
-private:
-  Obj_iterator *db_grants;  ///< database-level grants
-  Obj_iterator *tbl_grants; ///< table- and routine-level grants
-  Obj_iterator *col_grants; ///< column-level grants
 };
 
 ///////////////////////////////////////////////////////////////////////////
 
-// The functions in this section are intended to construct an instance of
-// Obj class for any particular database object. These functions do not
-// interact with the server to validate requested names. So, it is possible
-// to construct instances for non-existing objects.
+// Functions in this section are intended to construct an instance of Obj
+// class for any particular database object. These functions do not interact
+// with the server to validate requested names. So, it is possible to
+// construct instances for non-existing objects.
 //
 // The client is responsible for destroying the returned object.
 
 /**
   Construct an instance of Obj representing a database.
 
-  No actual actions are performed in the server. An object can be created
-  even for invalid database name or for non-existing database.
+  No actions are performed in the server. An object can be created
+  even for an invalid database name or for a non-existing database.
 
-  The client is responsible to destroy the created object.
+  The client is responsible for destruction of the created object.
 
   @param[in] db_name Database name.
 
-  @return a pointer to an instance of Obj representing given database.
-*/
-
-Obj *get_database(const String *db_name);
-
-/**
-  Construct an instance of Obj representing a table.
-
-  No actual actions are performed in the server. An object can be created
-  even for invalid database/table name or for non-existing table.
-
-  The client is responsible to destroy the created object.
-
-  @param[in] db_name    Database name.
-  @param[in] table_name Table name.
-
-  @return a pointer to an instance of Obj representing given table.
+  @return a pointer to an instance of Obj representing the given database.
 */
-
-Obj *get_table(const String *db_name, const String *table_name);
-
-/**
-  Construct an instance of Obj representing a view.
-
-  No actual actions are performed in the server. An object can be created
-  even for invalid database/view name or for non-existing view.
-
-  The client is responsible to destroy the created object.
-
-  @param[in] db_name   Database name.
-  @param[in] view_name View name.
-
-  @return a pointer to an instance of Obj representing given view.
-*/
-
-Obj *get_view(const String *db_name, const String *view_name);
-
-/**
-  Construct an instance of Obj representing a trigger.
-
-  No actual actions are performed in the server. An object can be created
-  even for invalid database/trigger name or for non-existing trigger.
-
-  The client is responsible to destroy the created object.
-
-  @param[in] db_name      Database name.
-  @param[in] trigger_name Trigger name.
-
-  @return a pointer to an instance of Obj representing given trigger.
-*/
-
-Obj *get_trigger(const String *db_name, const String *trigger_name);
-
-/**
-  Construct an instance of Obj representing a stored procedure.
-
-  No actual actions are performed in the server. An object can be created
-  even for invalid database/procedure name or for non-existing stored
-  procedure.
-
-  The client is responsible to destroy the created object.
-
-  @param[in] db_name Database name.
-  @param[in] sp_name Stored procedure name.
-
-  @return a pointer to an instance of Obj representing given stored
-  procedure.
-*/
-
-Obj *get_stored_procedure(const String *db_name, const String *sp_name);
-
-/**
-  Construct an instance of Obj representing a stored function.
-
-  No actual actions are performed in the server. An object can be created
-  even for invalid database/function name or for non-existing stored
-  function.
-
-  The client is responsible to destroy the created object.
-
-  @param[in] db_name Database name.
-  @param[in] sf_name Stored function name.
-
-  @return a pointer to an instance of Obj representing given stored
-  function.
-*/
-
-Obj *get_stored_function(const String *db_name, const String *sf_name);
-
-/**
-  Construct an instance of Obj representing an event.
-
-  No actual actions are performed in the server. An object can be created
-  even for invalid database/event name or for non-existing event.
-
-  The client is responsible to destroy the created object.
-
-  @param[in] db_name    Database name.
-  @param[in] event_name Event name.
-
-  @return a pointer to an instance of Obj representing given event.
-*/
-
-Obj *get_event(const String *db_name, const String *event_name);
+Obj *get_database_stub(const String *db_name);
 
 ///////////////////////////////////////////////////////////////////////////
 
-// The functions in this section provides a way to iterator over all
-// objects in the server or in the particular database.
+// Functions in this section provide a way to iterate over all objects in
+// the server or in a particular database.
 //
-// The client is responsible for destroying the returned iterator.
+// The client is responsible for destruction of the returned iterator.
+
+///////////////////////////////////////////////////////////////////////////
 
 /**
   Create an iterator over all databases in the server.
+  Includes system databases, such as "mysql" and "information_schema".
 
-  The client is responsible to destroy the returned iterator.
+  The client is responsible for destruction of the returned iterator.
 
   @return a pointer to an iterator object.
 */
-
 Obj_iterator *get_databases(THD *thd);
 
 /**
-  Create an iterator over all tables in the particular database.
+  Create an iterator over all base tables in a particular database.
+  Temporary tables are not included.
 
-  The client is responsible to destroy the returned iterator.
+  The client is responsible for destruction of the returned iterator.
 
   @return a pointer to an iterator object.
     @retval NULL in case of error.
 */
-
 Obj_iterator *get_db_tables(THD *thd, const String *db_name);
 
 /**
-  Create an iterator over all views in the particular database.
+  Create an iterator over all views in a particular database.
 
-  The client is responsible to destroy the returned iterator.
+  The client is responsible for destruction of the returned iterator.
 
   @return a pointer to an iterator object.
     @retval NULL in case of error.
 */
-
 Obj_iterator *get_db_views(THD *thd, const String *db_name);
 
 /**
-  Create an iterator over all triggers in the particular database.
+  Create an iterator over all triggers in a particular database.
 
-  The client is responsible to destroy the returned iterator.
+  The client is responsible for destruction of the returned iterator.
 
   @return a pointer to an iterator object.
     @retval NULL in case of error.
 */
-
 Obj_iterator *get_db_triggers(THD *thd, const String *db_name);
 
 /**
-  Create an iterator over all stored procedures in the particular database.
+  Create an iterator over all stored procedures in a particular database.
 
-  The client is responsible to destroy the returned iterator.
+  The client is responsible for destruction of the returned iterator.
 
   @return a pointer to an iterator object.
     @retval NULL in case of error.
 */
-
 Obj_iterator *get_db_stored_procedures(THD *thd, const String *db_name);
 
 /**
-  Create an iterator over all stored functions in the particular database.
+  Create an iterator over all stored functions in a particular database.
 
-  The client is responsible to destroy the returned iterator.
+  The client is responsible for destruction of the returned iterator.
 
   @return a pointer to an iterator object.
     @retval NULL in case of error.
 */
-
 Obj_iterator *get_db_stored_functions(THD *thd, const String *db_name);
 
 /**
-  Create an iterator over all events in the particular database.
+  Create an iterator over all events in a particular database.
 
-  The client is responsible to destroy the returned iterator.
+  The client is responsible for destruction of the returned iterator.
 
   @return a pointer to an iterator object.
     @retval NULL in case of error.
 */
-
 Obj_iterator *get_db_events(THD *thd, const String *db_name);
 
 /*
   Creates a high-level iterator that iterates over database-, table-,
-  routine-, and column-level privileges which shall permit a single
-  iterator from the si_objects to retrieve all of the privileges for
-  a given database.
+  routine-, and column-level- privileges. This allows to retrieve all
+  privileges of a given database using a single iterator.
 */
 Obj_iterator *get_all_db_grants(THD *thd, const String *db_name);
 
@@ -481,83 +244,92 @@ Obj_iterator *get_all_db_grants(THD *thd
 
 // The functions are intended to enumerate dependent objects.
 //
-// The client is responsible for destroying the returned iterator.
+// The client is responsible for destruction of the returned iterator.
+
+///////////////////////////////////////////////////////////////////////////
 
 /**
-  Create an iterator overl all base tables in the particular view.
+  Create an iterator over all base tables of a view.
 
-  The client is responsible to destroy the returned iterator.
+  The client is responsible for destruction of the returned iterator.
 
   @return a pointer to an iterator object.
     @retval NULL in case of error.
 */
-
 Obj_iterator* get_view_base_tables(THD *thd,
                                    const String *db_name,
                                    const String *view_name);
 
 /**
-  Create an iterator overl all base tables in the particular view.
+  Create an iterator over all base views of a particular view.
 
-  The client is responsible to destroy the returned iterator.
+  The client is responsible for destruction of the returned iterator.
 
   @return a pointer to an iterator object.
     @retval NULL in case of error.
 */
-
 Obj_iterator* get_view_base_views(THD *thd,
                                   const String *db_name,
                                   const String *view_name);
 
 ///////////////////////////////////////////////////////////////////////////
 
-// The functions in this section provides a way to materialize objects from
-// the serialized form.
+// Functions in this section provide a way to materialize objects from their
+// serialized form (serialization image). In order to do that, the client
+// creates an object handle, by means of one of the functions below, and
+// then calls "create()" method on it.
 //
-// The client is responsible for destroying the returned iterator.
+// The client is responsible for destruction of the returned object handle.
+//
+// The client is responsible for providing valid serialization image. The
+// operations below do not modify serialization image in any way. In
+// particular, the client is responsible to advance in the serialiazation
+// stream after successful object restoration.
+
+///////////////////////////////////////////////////////////////////////////
 
-Obj *materialize_database(const String *db_name,
-                          uint serialization_version,
-                          const String *serialization);
-
-Obj *materialize_table(const String *db_name,
-                       const String *table_name,
-                       uint serialization_version,
-                       const String *serialization);
-
-Obj *materialize_view(const String *db_name,
-                      const String *view_name,
-                      uint serialization_version,
-                      const String *serialization);
-
-Obj *materialize_trigger(const String *db_name,
-                         const String *trigger_name,
-                         uint serialization_version,
-                         const String *serialization);
-
-Obj *materialize_stored_procedure(const String *db_name,
-                                  const String *stored_proc_name,
-                                  uint serialization_version,
-                                  const String *serialization);
-
-Obj *materialize_stored_function(const String *db_name,
-                                 const String *stored_func_name,
-                                 uint serialization_version,
-                                 const String *serialization);
-
-Obj *materialize_event(const String *db_name,
-                       const String *event_name,
-                       uint serialization_version,
-                       const String *serialization);
-
-Obj *materialize_tablespace(const String *ts_name,
-                            uint serialization_version,
-                            const String *serialization);
-
-Obj *materialize_db_grant(const String *grantee,
-                          const String *db_name,
-                          uint serialization_version,
-                          const String *serialization);
+Obj *get_database(const String *db_name,
+                  uint image_version,
+                  const String *image);
+
+Obj *get_table(const String *db_name,
+               const String *table_name,
+               uint image_version,
+               const String *image);
+
+Obj *get_view(const String *db_name,
+              const String *view_name,
+              uint image_version,
+              const String *image);
+
+Obj *get_trigger(const String *db_name,
+                 const String *trigger_name,
+                 uint image_version,
+                 const String *image);
+
+Obj *get_stored_procedure(const String *db_name,
+                          const String *stored_proc_name,
+                          uint image_version,
+                          const String *image);
+
+Obj *get_stored_function(const String *db_name,
+                         const String *stored_func_name,
+                         uint image_version,
+                         const String *image);
+
+Obj *get_event(const String *db_name,
+               const String *event_name,
+               uint image_version,
+               const String *image);
+
+Obj *get_tablespace(const String *ts_name,
+                    uint image_version,
+                    const String *image);
+
+Obj *get_db_grant(const String *db_name,
+                  const String *name,
+                  uint image_version,
+                  const String *image);
 
 ///////////////////////////////////////////////////////////////////////////
 
@@ -565,14 +337,11 @@ Obj *materialize_db_grant(const String *
   Check if the given database name is reserved for internal use.
 
   @return
-    @retval TRUE if the given database name is reserved for internal use
+    @retval TRUE if the given database name is reserved for internal use.
     @retval FALSE otherwise.
 */
-
 bool is_internal_db_name(const String *db_name);
 
-///////////////////////////////////////////////////////////////////////////
-
 /**
   Check if the given directory actually exists.
 
@@ -580,42 +349,49 @@ bool is_internal_db_name(const String *d
     @retval FALSE on success (the database exists and accessible).
     @retval TRUE on error (the database either not exists, or not accessible).
 */
+bool check_db_existence(THD *thd, const String *db_name);
 
-bool check_db_existence(const String *db_name);
-
-/*
-  Check if the given user is actually defined on the system.
+/**
+  Check if the user is defined on the system.
 
-  @return Existence status.
-    @retval TRUE if user is defined on the system.
-    @retval FALSE if user does not exist.
+  @return
+    @retval TRUE if user is defined
+    @retval FALSE otherwise
 */
+bool check_user_existence(THD *thd, const Obj *obj);
 
-bool check_user_existence(THD *thd, const String *grantee);
+/**
+  Return user name of materialized grant object.
+*/
+const String *grant_get_user_name(const Obj *obj);
 
-/*
-  This method returns a @c TablespaceObj object if the table has a tablespace.
+/**
+  Return grant info of materialized grant object.
 */
+const String *grant_get_grant_info(const Obj *obj);
 
-Obj *get_tablespace_for_table(THD *thd,
-                              const String *db_name,
-                              const String *tbl_name);
+/**
+  Determine if the tablespace referenced by name exists on the system.
 
-/*
-  This method determines if a materialized tablespace exists on the
-  system. This compares the name and all saved attributes of the
-  tablespace. A FALSE return would mean either the tablespace does
-  not exist or the tablespace attributes are different.
+  @return a Tablespace_obj if it exists or NULL if it doesn't.
 */
+Obj *find_tablespace(THD *thd, const String *ts_name);
 
-bool tablespace_exists(THD *thd, Obj *ts);
-
-/*
-  This method determines if the tablespace referenced by name exists on the
-  system. Returns a TablespaceObj if it exists or NULL if it doesn't.
+/**
+  This method returns a @c Tablespace_obj object if the table has a
+  tablespace.
 */
+Obj *find_tablespace_for_table(THD *thd,
+                               const String *db_name,
+                               const String *table_name);
 
-Obj *is_tablespace(THD *thd, Obj *ts);
+/**
+  This method determines if a materialized tablespace exists on the system.
+  This compares the name and all saved attributes of the tablespace. A
+  FALSE return would mean either the tablespace does not exist or the
+  tablespace attributes are different.
+*/
+bool compare_tablespace_attributes(Obj *ts1, Obj *ts2);
 
 ///////////////////////////////////////////////////////////////////////////
 
@@ -623,6 +399,8 @@ Obj *is_tablespace(THD *thd, Obj *ts);
 // DDL blocker methods.
 //
 
+///////////////////////////////////////////////////////////////////////////
+
 /**
   Turn on the ddl blocker.
 
@@ -662,14 +440,6 @@ void ddl_blocker_exception_on(THD *thd);
   @param[in] thd  Thread context.
 */
 void ddl_blocker_exception_off(THD *thd);
-
-/*
-  Creates a WHERE clause for information schema table lookups of the
-  for FROM INFORMATION_SCHEMA.X WHERE <db_col> IN ('a','b','c').
-*/
-COND *create_db_select_condition(THD *thd,
-                                 TABLE *t,
-                                 List<LEX_STRING> *db_list);
 
 /*
   The following class is used to manage name locks on a list of tables.

=== modified file 'sql/sql_cache.cc'
--- a/sql/sql_cache.cc	2008-11-11 16:47:57 +0000
+++ b/sql/sql_cache.cc	2008-11-24 23:00:01 +0000
@@ -1014,7 +1014,9 @@ void Query_cache::store_query(THD *thd, 
       protocol (COM_EXECUTE) cannot be served to statements asking for results
       in the text protocol (COM_QUERY) and vice-versa.
     */
-    flags.result_in_binary_protocol= (unsigned int) thd->protocol->type();
+    flags.protocol_type= (unsigned int) thd->protocol->type();
+    /* PROTOCOL_LOCAL results are not cached. */
+    DBUG_ASSERT(flags.protocol_type != (unsigned int) Protocol::PROTOCOL_LOCAL);
     flags.more_results_exists= test(thd->server_status &
                                     SERVER_MORE_RESULTS_EXISTS);
     flags.pkt_nr= net->pkt_nr;
@@ -1041,7 +1043,7 @@ sql mode: 0x%lx, sort len: %lu, conncat 
 def_week_frmt: %lu",                          
                           (int)flags.client_long_flag,
                           (int)flags.client_protocol_41,
-                          (int)flags.result_in_binary_protocol,
+                          (int)flags.protocol_type,
                           (int)flags.more_results_exists,
                           flags.pkt_nr,
                           flags.character_set_client_num,
@@ -1279,7 +1281,7 @@ Query_cache::send_result_to_client(THD *
   flags.client_long_flag= test(thd->client_capabilities & CLIENT_LONG_FLAG);
   flags.client_protocol_41= test(thd->client_capabilities &
                                  CLIENT_PROTOCOL_41);
-  flags.result_in_binary_protocol= (unsigned int)thd->protocol->type();
+  flags.protocol_type= (unsigned int) thd->protocol->type();
   flags.more_results_exists= test(thd->server_status &
                                   SERVER_MORE_RESULTS_EXISTS);
   flags.pkt_nr= thd->net.pkt_nr;
@@ -1304,7 +1306,7 @@ sql mode: 0x%lx, sort len: %lu, conncat 
 def_week_frmt: %lu",                          
                           (int)flags.client_long_flag,
                           (int)flags.client_protocol_41,
-                          (int)flags.result_in_binary_protocol,
+                          (int)flags.protocol_type,
                           (int)flags.more_results_exists,
                           flags.pkt_nr,
                           flags.character_set_client_num,

=== modified file 'sql/sql_error.h'
--- a/sql/sql_error.h	2008-11-18 22:30:59 +0000
+++ b/sql/sql_error.h	2008-11-19 19:20:47 +0000
@@ -80,8 +80,6 @@ private:
 public:
 
   Warning_info(ulonglong warn_id_arg);
-  /* Do nothing - used to initialize a backup info */
-  Warning_info() { clear_alloc_root(&m_warn_root); }
   ~Warning_info();
 
   /**

=== modified file 'sql/sql_lex.cc'
--- a/sql/sql_lex.cc	2008-10-16 02:13:16 +0000
+++ b/sql/sql_lex.cc	2008-11-24 23:00:01 +0000
@@ -37,6 +37,7 @@ sys_var *trg_new_row_fake_var= (sys_var*
   LEX_STRING constant for null-string to be used in parser and other places.
 */
 const LEX_STRING null_lex_str= {NULL, 0};
+const LEX_STRING empty_lex_str= { (char*) "", 0 };
 
 /* Longest standard keyword name */
 
@@ -144,6 +145,7 @@ Lex_input_stream::Lex_input_stream(THD *
   found_semicolon(NULL),
   ignore_space(test(thd->variables.sql_mode & MODE_IGNORE_SPACE)),
   stmt_prepare_mode(FALSE),
+  multi_statements(TRUE),
   in_comment(NO_COMMENT),
   m_underscore_cs(NULL)
 {

=== modified file 'sql/sql_lex.h'
--- a/sql/sql_lex.h	2008-11-19 18:34:03 +0000
+++ b/sql/sql_lex.h	2008-11-24 23:00:01 +0000
@@ -942,6 +942,7 @@ enum xa_option_words {XA_NONE, XA_JOIN, 
                       XA_SUSPEND, XA_FOR_MIGRATE};
 
 extern const LEX_STRING null_lex_str;
+extern const LEX_STRING empty_lex_str;
 
 
 /*
@@ -1484,7 +1485,6 @@ public:
     TRUE if we should allow multi-statements.
   */
   bool multi_statements;
-
 
   /** State of the lexical analyser for comments. */
   enum_comment_state in_comment;

=== modified file 'sql/sql_prepare.cc'
--- a/sql/sql_prepare.cc	2008-11-19 18:34:03 +0000
+++ b/sql/sql_prepare.cc	2008-11-28 21:35:11 +0000
@@ -162,7 +162,7 @@ public:
   bool execute_loop(String *expanded_query,
                     bool open_cursor,
                     uchar *packet_arg, uchar *packet_end_arg);
-  bool execute_immediate(const char *stmt, uint length);
+  bool execute_server_runnable(Server_runnable *server_runnable);
   /* Destroy this statement */
   void deallocate();
 private:
@@ -184,6 +184,18 @@ private:
 };
 
 
+/**
+*/
+
+class Execute_sql_statement: public Server_runnable
+{
+public:
+  Execute_sql_statement(LEX_STRING sql_text);
+  virtual bool execute_server_code(THD *thd);
+private:
+  LEX_STRING m_sql_text;
+};
+
 /******************************************************************************
   Implementation
 ******************************************************************************/
@@ -2764,6 +2776,54 @@ void mysql_stmt_get_longdata(THD *thd, c
 }
 
 
+bool
+mysql_execute_direct(THD *thd, LEX_STRING query, Ed_result *ed_result)
+{
+  Execute_sql_statement execute_sql_statement(query);
+
+  return mysql_execute_direct(thd, &execute_sql_statement, ed_result);
+}
+
+
+/**
+  Execute a fragment of server functionality without an effect on
+  thd, and store results in Ed_result.
+
+  @param thd         Thread handle.
+  @param server_runnable A code fragment to execute.
+  @param ed_result   Result interceptor
+*/
+
+bool
+mysql_execute_direct(THD *thd, Server_runnable *server_runnable,
+                     Ed_result *ed_result)
+{
+  Protocol_local protocol_local(thd, ed_result);
+  Prepared_statement stmt(thd);
+
+  DBUG_ENTER("mysql_execute_direct");
+
+  DBUG_ASSERT(ed_result);
+
+  Protocol *protocol_saved= thd->protocol;
+
+  thd->protocol= &protocol_local;
+
+  ed_result->begin_statement(thd);
+  bool rc= stmt.execute_server_runnable(server_runnable);
+  ed_result->end_statement(thd);
+
+  thd->protocol->end_statement();
+
+  thd->protocol= protocol_saved;
+
+  thd->main_da.reset_diagnostics_area();
+
+  DBUG_RETURN(rc);
+}
+
+
+
 /***************************************************************************
  Select_fetch_protocol_binary
 ****************************************************************************/
@@ -2809,6 +2869,63 @@ Select_fetch_protocol_binary::send_data(
   return rc;
 }
 
+/*******************************************************************
+*
+*******************************************************************/
+
+Server_runnable::~Server_runnable()
+{
+}
+
+///////////////////////////////////////////////////////////////////////////
+
+Execute_sql_statement::
+Execute_sql_statement(LEX_STRING sql_text)
+  :m_sql_text(sql_text)
+{}
+
+
+/**
+  Parse and execute a statement. Does not prepare the query.
+
+  Allows to execute a statement from within another statement.
+  The main property of the implementation is that it does not
+  affect the environment -- i.e. you  can run many
+  executions without having to cleanup/reset THD in between.
+*/
+
+bool
+Execute_sql_statement::execute_server_code(THD *thd)
+{
+  bool error;
+
+  if (alloc_query(thd, m_sql_text.str, m_sql_text.length))
+    return TRUE;
+
+  Parser_state parser_state(thd, thd->query, thd->query_length);
+
+  parser_state.m_lip.multi_statements= FALSE;
+  lex_start(thd);
+
+  error= parse_sql(thd, &parser_state, NULL) || thd->is_error();
+
+  if (error)
+    goto end;
+
+  thd->lex->set_trg_event_type_for_tables();
+
+  error= mysql_execute_command(thd);
+
+  if (error == 0 && thd->spcont == NULL)
+    general_log_write(thd, COM_STMT_EXECUTE,
+                      thd->query, thd->query_length);
+
+end:
+  lex_end(thd->lex);
+
+  return error;
+}
+
 /***************************************************************************
  Prepared_statement
 ****************************************************************************/
@@ -3259,81 +3376,42 @@ reexecute:
 }
 
 
-/**
-  Parse and execute a query string. Does not prepare the query.
-
-  An implementation of "EXECUTE IMMEDIATE" syntax of Dynamic SQL.
-  Allows to execute a statement from within another statement.
-  The main property of the implementation is that it does not
-  affect the environment -- i.e. you  can run many
-  ::execute_immediate() without having to cleanup/reset THD in
-  between.
-*/
-
 bool
-Prepared_statement::execute_immediate(const char *query, uint length)
+Prepared_statement::execute_server_runnable(Server_runnable *server_runnable)
 {
   Statement stmt_backup;
   bool error;
+  Query_arena *save_stmt_arena= thd->stmt_arena;
+  Item_change_list save_change_list= thd->change_list;
 
   state= CONVENTIONAL_EXECUTION;
 
-  if (!(lex = new (mem_root) st_lex_local))
+  if (!(lex= new (mem_root) st_lex_local))
     return TRUE;
 
   thd->set_n_backup_statement(this, &stmt_backup);
+  thd->set_n_backup_active_arena(this, &stmt_backup);
+  thd->stmt_arena= this;
+  thd->change_list.empty();
 
-  if (alloc_query(thd, query, length))
-  {
-    thd->set_statement(&stmt_backup);
-    return TRUE;
-  }
-  /*
-    Expanded query is needed for slow logging, so we want thd->query
-    to point at it even after we restore from backup. This is ok, as
-    expanded query was allocated in thd->mem_root.
-  */
-  stmt_backup.query= thd->query;
-  stmt_backup.query_length= thd->query_length;
-
-  Parser_state parser_state(thd, thd->query, thd->query_length);
-
-  parser_state.m_lip.multi_statements= FALSE;
-  lex_start(thd);
-
-  error= parse_sql(thd, &parser_state, NULL) || thd->is_error();
-
-
-  if (lex->sql_command == SQLCOM_PREPARE ||
-      lex->sql_command == SQLCOM_EXECUTE ||
-      (lex->sql_command == SQLCOM_CHANGE_DB && is_sql_prepare()))
-  {
-    my_error(ER_PS_NO_RECURSION, MYF(0));
-    error= 1;
-  }
-
-  if (error)
-  {
-    thd->set_statement(&stmt_backup);
-    return TRUE;
-  }
-
-  lex->set_trg_event_type_for_tables();
+  error= server_runnable->execute_server_code(thd);
 
-  if (query_cache_send_result_to_client(thd, thd->query,
-                                        thd->query_length) <= 0)
-  {
-    error= mysql_execute_command(thd);
-  }
+  delete lex->sphead;
+  lex->sphead= 0;
+  /* The order is important */
+  lex->unit.cleanup();
+  close_thread_tables(thd);
+  thd->cleanup_after_query();
 
-  cleanup_stmt();
-  lex_end(thd->lex);
+  thd->restore_active_arena(this, &stmt_backup);
+  thd->restore_backup_statement(this, &stmt_backup);
+  thd->stmt_arena= save_stmt_arena;
 
-  thd->set_statement(&stmt_backup);
+  DBUG_ASSERT(thd->change_list.is_empty());
+  thd->change_list= save_change_list;
+  save_change_list.empty();
 
-  if (error == 0 && thd->spcont == NULL)
-    general_log_write(thd, COM_STMT_EXECUTE,
-                      thd->query, thd->query_length);
+  /* Items and memory will freed in destructor */
 
   return error;
 }

=== modified file 'sql/sql_string.h'
--- a/sql/sql_string.h	2008-08-20 10:29:58 +0000
+++ b/sql/sql_string.h	2008-11-24 23:00:01 +0000
@@ -113,6 +113,11 @@ public:
       (void) realloc(str_length);
     return Ptr;
   }
+  LEX_STRING lex_string() const
+  {
+    LEX_STRING lex_string = { (char*) ptr(), length() };
+    return lex_string;
+  }
 
   void set(String &str,uint32 offset,uint32 arg_length)
   {

=== modified file 'sql/sql_yacc.yy'
--- a/sql/sql_yacc.yy	2008-11-06 18:39:27 +0000
+++ b/sql/sql_yacc.yy	2008-11-19 16:32:01 +0000
@@ -1485,7 +1485,7 @@ query:
             Lex_input_stream *lip = YYLIP;
 
             if ((YYTHD->client_capabilities & CLIENT_MULTI_QUERIES) &&
-                ! lip->stmt_prepare_mode &&
+                lip->multi_statements &&
                 ! lip->eof())
             {
               /*

Thread
bzr commit into mysql-6.0-runtime branch (alik:2756) WL#4264Alexander Nozdrin2 Dec