Merge mysql.com:/users/lthalmann/bkroot/mysql-5.1-new-rpl

into  mysql.com:/users/lthalmann/bk/MERGE/mysql-5.1-merge


mysql-test/t/archive.test:
  Auto merged
sql/log.cc:
  Auto merged
sql/sql_base.cc:
  Auto merged
sql/sql_class.cc:
  Auto merged
sql/sql_class.h:
  Auto merged
sql/sql_insert.cc:
  Auto merged
sql/sql_parse.cc:
  Auto merged
sql/sql_table.cc:
  Auto merged

11 files changed, 427 insertions, 194 deletions

diff --git a/sql/ha_innodb.cc b/sql/ha_innodb.cc
index 742f9ce7631..ec504cc1266 100644
--- a/sql/ha_innodb.cc
+++ b/sql/ha_innodb.cc
@@ -135,6 +135,7 @@ extern "C" {
 #include "../storage/innobase/include/fil0fil.h"
 #include "../storage/innobase/include/trx0xa.h"
 #include "../storage/innobase/include/thr0loc.h"
+#include "../storage/innobase/include/ha_prototypes.h"
 }
 
 #define HA_INNOBASE_ROWS_IN_TABLE 10000 /* to get optimization right */
@@ -668,6 +669,61 @@ innobase_get_cset_width(
 }
 
 /**********************************************************************
+Converts an identifier to a table name.
+
+NOTE that the exact prototype of this function has to be in
+/innobase/dict/dict0dict.c! */
+extern "C"
+void
+innobase_convert_from_table_id(
+/*===========================*/
+	char*		to,	/* out: converted identifier */
+	const char*	from,	/* in: identifier to convert */
+	ulint		len)	/* in: length of 'to', in bytes */
+{
+	uint	errors;
+
+	strconvert(current_thd->charset(), from,
+		   &my_charset_filename, to, len, &errors);
+}
+
+/**********************************************************************
+Converts an identifier to UTF-8.
+
+NOTE that the exact prototype of this function has to be in
+/innobase/dict/dict0dict.c! */
+extern "C"
+void
+innobase_convert_from_id(
+/*=====================*/
+	char*		to,	/* out: converted identifier */
+	const char*	from,	/* in: identifier to convert */
+	ulint		len)	/* in: length of 'to', in bytes */
+{
+	uint	errors;
+
+	strconvert(current_thd->charset(), from,
+		   system_charset_info, to, len, &errors);
+}
+
+/**********************************************************************
+Removes the filename encoding of a table or database name.
+
+NOTE that the exact prototype of this function has to be in
+/innobase/dict/dict0dict.c! */
+extern "C"
+void
+innobase_convert_from_filename(
+/*===========================*/
+	char*		s)	/* in: identifier; out: decoded identifier */
+{
+	uint	errors;
+
+	strconvert(&my_charset_filename, s,
+		   system_charset_info, s, strlen(s), &errors);
+}
+
+/**********************************************************************
 Compares NUL-terminated UTF-8 strings case insensitively.
 
 NOTE that the exact prototype of this function has to be in
@@ -697,6 +753,21 @@ innobase_casedn_str(
 	my_casedn_str(system_charset_info, a);
 }
 
+/**************************************************************************
+Determines the connection character set.
+
+NOTE that the exact prototype of this function has to be in
+/innobase/dict/dict0dict.c! */
+extern "C"
+struct charset_info_st*
+innobase_get_charset(
+/*=================*/
+				/* out: connection character set */
+	void*	mysql_thd)	/* in: MySQL thread handle */
+{
+	return(((THD*) mysql_thd)->charset());
+}
+
 /*************************************************************************
 Creates a temporary file. */
 extern "C"
@@ -743,6 +814,25 @@ innobase_mysql_tmpfile(void)
 }
 
 /*************************************************************************
+Wrapper around MySQL's copy_and_convert function, see it for
+documentation. */
+extern "C"
+ulint
+innobase_convert_string(
+/*====================*/
+	void*		to,
+	ulint		to_length,
+	CHARSET_INFO*	to_cs,
+	const void*	from,
+	ulint		from_length,
+	CHARSET_INFO*	from_cs,
+	uint*		errors)
+{
+	return(copy_and_convert((char*)to, to_length, to_cs,
+		       (const char*)from, from_length, from_cs, errors));
+}
+
+/*************************************************************************
 Gets the InnoDB transaction handle for a MySQL handler object, creates
 an InnoDB transaction struct if the corresponding MySQL thread struct still
 lacks one. */
@@ -1076,23 +1166,69 @@ innobase_invalidate_query_cache(
 }
 
 /*********************************************************************
-Get the quote character to be used in SQL identifiers.
+Display an SQL identifier.
 This definition must match the one in innobase/ut/ut0ut.c! */
 extern "C"
-int
-mysql_get_identifier_quote_char(
-/*============================*/
-				/* out: quote character to be
-				used in SQL identifiers; EOF if none */
+void
+innobase_print_identifier(
+/*======================*/
+	FILE*		f,	/* in: output stream */
 	trx_t*		trx,	/* in: transaction */
+	ibool		table_id,/* in: TRUE=decode table name */
 	const char*	name,	/* in: name to print */
 	ulint		namelen)/* in: length of name */
 {
+	const char*	s	= name;
+	char*		qname	= NULL;
+	int		q;
+
+	if (table_id) {
+		/* Decode the table name.  The filename_to_tablename()
+		function expects a NUL-terminated string.  The input and
+		output strings buffers must not be shared.  The function
+		only produces more output when the name contains other
+		characters than [0-9A-Z_a-z]. */
+		char*	temp_name = my_malloc(namelen + 1, MYF(MY_WME));
+		uint	qnamelen = namelen
+				+ (1 + sizeof srv_mysql50_table_name_prefix);
+
+		if (temp_name) {
+			qname = my_malloc(qnamelen, MYF(MY_WME));
+			if (qname) {
+				memcpy(temp_name, name, namelen);
+				temp_name[namelen] = 0;
+				s = qname;
+				namelen = filename_to_tablename(temp_name,
+						qname, qnamelen);
+			}
+			my_free(temp_name, MYF(0));
+		}
+	}
+
 	if (!trx || !trx->mysql_thd) {
-		return(EOF);
+
+		q = '"';
+	} else {
+		q = get_quote_char_for_identifier((THD*) trx->mysql_thd,
+						s, (int) namelen);
+	}
+
+	if (q == EOF) {
+		fwrite(s, 1, namelen, f);
+	} else {
+		const char*	e = s + namelen;
+		putc(q, f);
+		while (s < e) {
+			int	c = *s++;
+			if (c == q) {
+				putc(c, f);
+			}
+			putc(c, f);
+		}
+		putc(q, f);
 	}
-	return(get_quote_char_for_identifier((THD*) trx->mysql_thd,
-						name, (int) namelen));
+
+	my_free(qname, MYF(MY_ALLOW_ZERO_PTR));
 }
 
 /**************************************************************************
@@ -1232,6 +1368,24 @@ innobase_init(void)
 
 	ut_a(DATA_MYSQL_TRUE_VARCHAR == (ulint)MYSQL_TYPE_VARCHAR);
 
+#ifdef UNIV_DEBUG
+	static const char	test_filename[] = "-@";
+	char			test_tablename[sizeof test_filename
+				+ sizeof srv_mysql50_table_name_prefix];
+	if ((sizeof test_tablename) - 1
+			!= filename_to_tablename(test_filename, test_tablename,
+			sizeof test_tablename)
+			|| strncmp(test_tablename,
+			srv_mysql50_table_name_prefix,
+			sizeof srv_mysql50_table_name_prefix)
+			|| strcmp(test_tablename
+			+ sizeof srv_mysql50_table_name_prefix,
+			test_filename)) {
+		sql_print_error("tablename encoding has been changed");
+		goto error;
+	}
+#endif /* UNIV_DEBUG */
+
 	/* Check that values don't overflow on 32-bit systems. */
 	if (sizeof(ulint) == 4) {
 		if (innobase_buffer_pool_size > UINT_MAX32) {
@@ -2200,8 +2354,7 @@ ha_innobase::open(
 
 	/* Get pointer to a table object in InnoDB dictionary cache */
 
-	ib_table = dict_table_get_and_increment_handle_count(
-		norm_name, NULL);
+	ib_table = dict_table_get_and_increment_handle_count(norm_name);
 
 	if (NULL == ib_table) {
 		ut_print_timestamp(stderr);
@@ -4116,6 +4269,9 @@ ha_innobase::index_prev(
 	mysql_byte*	buf)	/* in/out: buffer for previous row in MySQL
 				format */
 {
+	statistic_increment(current_thd->status_var.ha_read_prev_count,
+		&LOCK_status);
+
 	return(general_fetch(buf, ROW_SEL_PREV, 0));
 }
 
@@ -4652,7 +4808,7 @@ ha_innobase::create(
 	/* Get the transaction associated with the current thd, or create one
 	if not yet created */
 
-	parent_trx = check_trx_exists(current_thd);
+	parent_trx = check_trx_exists(thd);
 
 	/* In case MySQL calls this in the middle of a SELECT query, release
 	possible adaptive hash latch to avoid deadlocks of threads */
@@ -4748,20 +4904,9 @@ ha_innobase::create(
 		}
 	}
 
-	if (current_thd->query != NULL) {
-		LEX_STRING q;
-
-		if (thd->convert_string(&q, system_charset_info,
-					current_thd->query,
-					current_thd->query_length,
-					current_thd->charset())) {
-			error = HA_ERR_OUT_OF_MEM;
-
-			goto cleanup;
-		}
-
+	if (thd->query != NULL) {
 		error = row_table_add_foreign_constraints(trx,
-			q.str, norm_name,
+			thd->query, norm_name,
 			create_info->options & HA_LEX_CREATE_TMP_TABLE);
 
 		error = convert_error_code_to_mysql(error, NULL);
@@ -4781,7 +4926,7 @@ ha_innobase::create(
 
 	log_buffer_flush_to_disk();
 
-	innobase_table = dict_table_get(norm_name, NULL);
+	innobase_table = dict_table_get(norm_name);
 
 	DBUG_ASSERT(innobase_table != 0);
 
@@ -4917,7 +5062,7 @@ ha_innobase::delete_table(
 	/* Get the transaction associated with the current thd, or create one
 	if not yet created */
 
-	parent_trx = check_trx_exists(current_thd);
+	parent_trx = check_trx_exists(thd);
 
 	/* In case MySQL calls this in the middle of a SELECT query, release
 	possible adaptive hash latch to avoid deadlocks of threads */
diff --git a/sql/handler.cc b/sql/handler.cc
index 47bcf1caba1..0895c6cf454 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -53,6 +53,8 @@ st_plugin_int *hton2plugin[MAX_HA];
 
 static handlerton *installed_htons[128];
 
+#define BITMAP_STACKBUF_SIZE (128/8)
+
 KEY_CREATE_INFO default_key_create_info= { HA_KEY_ALG_UNDEF, 0, {NullS,0} };
 
 /* static functions defined in this file */
@@ -3234,35 +3236,128 @@ namespace {
   }
 }
 
-template<class RowsEventT> int binlog_log_row(TABLE *table,
-                                              const byte *before_record,
-                                              const byte *after_record)
+/*
+   Write table maps for all (manually or automatically) locked tables
+   to the binary log.
+
+   SYNOPSIS
+     write_locked_table_maps()
+       thd     Pointer to THD structure
+
+   DESCRIPTION
+       This function will generate and write table maps for all tables
+       that are locked by the thread 'thd'.  Either manually locked
+       (stored in THD::locked_tables) and automatically locked (stored
+       in THD::lock) are considered.
+   
+   RETURN VALUE
+       0   All OK
+       1   Failed to write all table maps
+
+   SEE ALSO
+       THD::lock
+       THD::locked_tables
+ */
+namespace
 {
-  if (table->file->is_injective())
-    return 0;
-  bool error= 0;
+  int write_locked_table_maps(THD *thd)
+  {
+    DBUG_ENTER("write_locked_table_maps");
+    DBUG_PRINT("enter", ("thd=%p, thd->lock=%p, thd->locked_tables=%p",
+                         thd, thd->lock, thd->locked_tables));
+
+    if (thd->get_binlog_table_maps() == 0)
+    {
+      /*
+        Exactly one table has to be locked, otherwise this code is not
+        guaranteed to work.
+      */
+      DBUG_ASSERT((thd->lock != NULL) + (thd->locked_tables != NULL) == 1);
+
+      MYSQL_LOCK *lock= thd->lock ? thd->lock : thd->locked_tables;
+      DBUG_ASSERT(lock->table_count > 0);
+      TABLE **const end_ptr= lock->table + lock->table_count;
+      for (TABLE **table_ptr= lock->table ; 
+           table_ptr != end_ptr ;
+           ++table_ptr)
+      {
+        TABLE *const table= *table_ptr;
+        DBUG_PRINT("info", ("Checking table %s", table->s->table_name));
+        if (table->current_lock == F_WRLCK &&
+            check_table_binlog_row_based(thd, table))
+        {
+          int const has_trans= table->file->has_transactions();
+          int const error= thd->binlog_write_table_map(table, has_trans);
+          /*
+            If an error occurs, it is the responsibility of the caller to
+            roll back the transaction.
+          */
+          if (unlikely(error))
+            DBUG_RETURN(1);
+        }
+      }
+    }
+    DBUG_RETURN(0);
+  }
 
-  if (check_table_binlog_row_based(table->in_use, table))
+  template<class RowsEventT> int
+  binlog_log_row(TABLE* table,
+                 const byte *before_record,
+                 const byte *after_record)
   {
-    error=
-      RowsEventT::binlog_row_logging_function(table->in_use, table,
-                                              table->file->has_transactions(),
-                                              &table->s->all_set,
-                                              table->s->fields,
-                                              before_record, after_record);
+    if (table->file->is_injective())
+      return 0;
+    bool error= 0;
+    THD *const thd= table->in_use;
+
+    if (check_table_binlog_row_based(thd, table))
+    {
+      MY_BITMAP cols;
+      /* Potential buffer on the stack for the bitmap */
+      uint32 bitbuf[BITMAP_STACKBUF_SIZE/sizeof(uint32)];
+      uint n_fields= table->s->fields;
+      my_bool use_bitbuf= n_fields <= sizeof(bitbuf)*8;
+
+      /*
+        If there are no table maps written to the binary log, this is
+        the first row handled in this statement. In that case, we need
+        to write table maps for all locked tables to the binary log.
+      */
+      if (likely(!(error= bitmap_init(&cols,
+                                      use_bitbuf ? bitbuf : NULL,
+                                      (n_fields + 7) & ~7UL,
+                                      false))))
+      {
+        bitmap_set_all(&cols);
+        if (likely(!(error= write_locked_table_maps(thd))))
+        {
+          error=
+            RowsEventT::binlog_row_logging_function(thd, table,
+                                                    table->file->has_transactions(),
+                                                    &cols, table->s->fields,
+                                                    before_record, after_record);
+        }
+        if (!use_bitbuf)
+          bitmap_free(&cols);
+      }
+    }
+    return error ? HA_ERR_RBR_LOGGING_FAILED : 0;
   }
-  return error ? HA_ERR_RBR_LOGGING_FAILED : 0;
-}
 
+  /*
+    Instantiate the versions we need for the above template function,
+    because we have -fno-implicit-template as compiling option.
+  */
 
-/*
-  Instantiate the versions we need for the above template function, because we
-  have -fno-implicit-template as compiling option.
-*/
+  template int
+  binlog_log_row<Write_rows_log_event>(TABLE *, const byte *, const byte *);
+
+  template int
+  binlog_log_row<Delete_rows_log_event>(TABLE *, const byte *, const byte *);
 
-template int binlog_log_row<Write_rows_log_event>(TABLE *, const byte *, const byte *);
-template int binlog_log_row<Delete_rows_log_event>(TABLE *, const byte *, const byte *);
-template int binlog_log_row<Update_rows_log_event>(TABLE *, const byte *, const byte *);
+  template int
+  binlog_log_row<Update_rows_log_event>(TABLE *, const byte *, const byte *);
+}
 
 #endif /* HAVE_ROW_BASED_REPLICATION */
 
@@ -3272,41 +3367,6 @@ int handler::ha_external_lock(THD *thd, int lock_type)
   int error;
   if (unlikely(error= external_lock(thd, lock_type)))
     DBUG_RETURN(error);
-#ifdef HAVE_ROW_BASED_REPLICATION
-  if (table->file->is_injective())
-    DBUG_RETURN(0);
-
-  /*
-    There is a number of statements that are logged statement-based
-    but call external lock. For these, we do not need to generate a
-    table map.
-
-    TODO: The need for this switch is an indication that the model for
-    locking combined with row-based replication needs to be looked
-    over. Ideally, no such special handling should be needed.
-   */
-  switch (thd->lex->sql_command) {
-  case SQLCOM_TRUNCATE:
-  case SQLCOM_ALTER_TABLE:
-  case SQLCOM_OPTIMIZE:
-  case SQLCOM_REPAIR:
-    DBUG_RETURN(0);
-  default:
-    break;
-  }
-
-  /*
-    If we are locking a table for writing, we generate a table map.
-    For all other kinds of locks, we don't do anything.
-   */
-  if (lock_type == F_WRLCK && check_table_binlog_row_based(thd, table))
-  {
-    int const has_trans= table->file->has_transactions();
-    error= thd->binlog_write_table_map(table, has_trans);
-    if (unlikely(error))
-      DBUG_RETURN(error);
-  }
-#endif
   DBUG_RETURN(0);
 }
 
diff --git a/sql/log.cc b/sql/log.cc
index d05a8e4202f..11ee2dd58c9 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -71,11 +71,17 @@ sql_print_message_func sql_print_message_handlers[3] =
 struct binlog_trx_data {
   bool empty() const
   {
+#ifdef HAVE_ROW_BASED_REPLICATION
     return pending == NULL && my_b_tell(&trans_log) == 0;
+#else
+    return my_b_tell(&trans_log) == 0;
+#endif
   }
   binlog_trx_data() {}
   IO_CACHE trans_log;                         // The transaction cache
+#ifdef HAVE_ROW_BASED_REPLICATION
   Rows_log_event *pending;                // The pending binrows event
+#endif
 };
 
 handlerton binlog_hton;
@@ -1073,6 +1079,7 @@ binlog_end_trans(THD *thd, binlog_trx_data *trx_data, Log_event *end_ev)
 #endif
     error= mysql_bin_log.write(thd, trans_log, end_ev);
   }
+#ifdef HAVE_ROW_BASED_REPLICATION
   else
   {
 #ifdef HAVE_ROW_BASED_REPLICATION
@@ -1092,6 +1099,7 @@ binlog_end_trans(THD *thd, binlog_trx_data *trx_data, Log_event *end_ev)
     transaction cache.
   */
   mysql_bin_log.update_table_map_version();
+#endif
 
   statistic_increment(binlog_cache_use, &LOCK_status);
   if (trans_log->disk_writes != 0)
@@ -2919,7 +2927,6 @@ bool MYSQL_BIN_LOG::is_query_in_union(THD *thd, query_id_t query_id_param)
 }
 
 
-#ifdef HAVE_ROW_BASED_REPLICATION
 /*
   These functions are placed in this file since they need access to
   binlog_hton, which has internal linkage.
@@ -2955,6 +2962,7 @@ int THD::binlog_setup_trx_data()
   engine has registered for the transaction.
  */
 
+#ifdef HAVE_ROW_BASED_REPLICATION
 int THD::binlog_write_table_map(TABLE *table, bool is_trans)
 {
   int error;
@@ -3133,9 +3141,9 @@ bool MYSQL_BIN_LOG::write(Log_event *event_info)
     we are inside a stored function, we do not end the statement since
     this will close all tables on the slave.
   */
+#ifdef HAVE_ROW_BASED_REPLICATION
   bool const end_stmt=
     thd->prelocked_mode && thd->lex->requires_prelocking();
-#ifdef HAVE_ROW_BASED_REPLICATION
   thd->binlog_flush_pending_rows_event(end_stmt);
 #endif /*HAVE_ROW_BASED_REPLICATION*/
 
@@ -3186,7 +3194,6 @@ bool MYSQL_BIN_LOG::write(Log_event *event_info)
         (binlog_trx_data*) thd->ha_data[binlog_hton.slot];
       IO_CACHE *trans_log= &trx_data->trans_log;
       bool trans_log_in_use= my_b_tell(trans_log) != 0;
-
       if (event_info->get_cache_stmt() && !trans_log_in_use)
         trans_register_ha(thd,
                           (thd->options &
diff --git a/sql/log_event.cc b/sql/log_event.cc
index 78ab54aeb79..bf876366879 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -5595,6 +5595,7 @@ int Rows_log_event::exec_event(st_relay_log_info *rli)
 bool Rows_log_event::write_data_header(IO_CACHE *file)
 {
   byte buf[ROWS_HEADER_LEN];	// No need to init the buffer
+  DBUG_ASSERT(m_table_id != ~0UL);
   DBUG_EXECUTE_IF("old_row_based_repl_4_byte_map_id_master",
                   {
                     int4store(buf + 0, m_table_id);
@@ -5845,9 +5846,7 @@ int Table_map_log_event::exec_event(st_relay_log_info *rli)
 
     /*
       Open the table if it is not already open and add the table to table map.
-      If the table should not be replicated, we don't bother to do anything.
-      The table map will return NULL and the row-level event will effectively
-      be a no-op.
+      Note that for any table that should not be replicated, a filter is needed.
     */
     uint count;
     /*
@@ -5863,34 +5862,14 @@ int Table_map_log_event::exec_event(st_relay_log_info *rli)
         /*
           Error reporting borrowed from Query_log_event with many excessive
           simplifications (we don't honour --slave-skip-errors)
-
-          BUG: There can be extreneous table maps in the binary log,
-          so in case we fail to open the table, we just generate a
-          warning and do not add the table to the list of tables to
-          open and lock.
         */
         uint actual_error= thd->net.last_errno;
-        switch (actual_error)
-        {
-        case ER_NO_SUCH_TABLE:
-          slave_print_msg(WARNING_LEVEL, rli, actual_error,
-                          thd->net.last_error ?
-                          thd->net.last_error :
-                          "<no message>");
-          clear_all_errors(thd, rli);
-          rli->inc_event_relay_log_pos();       // Continue with next event
-          error= 0;
-          break;
-
-        default:
-          slave_print_msg(ERROR_LEVEL, rli, actual_error,
-                          "Error '%s' on opening table `%s`.`%s`",
-                          (actual_error ? thd->net.last_error :
-                           "unexpected success or fatal error"),
-                          table_list->db, table_list->table_name);
-          thd->query_error= 1;
-          break;
-        }
+        slave_print_msg(ERROR_LEVEL, rli, actual_error,
+                        "Error '%s' on opening table `%s`.`%s`",
+                        (actual_error ? thd->net.last_error :
+                         "unexpected success or fatal error"),
+                        table_list->db, table_list->table_name);
+        thd->query_error= 1;
       }
       DBUG_RETURN(error);
     }
diff --git a/sql/set_var.cc b/sql/set_var.cc
index a44395c74ca..acd17a07f78 100644
--- a/sql/set_var.cc
+++ b/sql/set_var.cc
@@ -1301,15 +1301,20 @@ bool sys_var_thd_binlog_format::is_readonly() const
     if global or not here.
     And this test will also prevent switching from RBR to RBR (a no-op which
     should not happen too often).
+
+    If we don't have row-based replication compiled in, the variable
+    is always read-only.
   */
-#ifdef HAVE_ROW_BASED_REPLICATION
+#ifndef HAVE_ROW_BASED_REPLICATION
+  my_error(ER_RBR_NOT_AVAILABLE, MYF(0));
+  return 1;
+#else
   if ((thd->variables.binlog_format == BINLOG_FORMAT_ROW) &&
       thd->temporary_tables)
   {
     my_error(ER_TEMP_TABLE_PREVENTS_SWITCH_OUT_OF_RBR, MYF(0));
     return 1;
   }
-#endif /*HAVE_ROW_BASED_REPLICATION*/
   /*
     if in a stored function, it's too late to change mode
   */
@@ -1327,10 +1332,12 @@ bool sys_var_thd_binlog_format::is_readonly() const
     my_error(ER_NDB_CANT_SWITCH_BINLOG_FORMAT, MYF(0));
     return 1;
   }
-#endif
+#endif /* HAVE_NDB_BINLOG */
+#endif /* HAVE_ROW_BASED_REPLICATION */
   return sys_var_thd_enum::is_readonly();
 }
 
+
 void fix_binlog_format_after_update(THD *thd, enum_var_type type)
 {
 #ifdef HAVE_ROW_BASED_REPLICATION
@@ -1338,6 +1345,7 @@ void fix_binlog_format_after_update(THD *thd, enum_var_type type)
 #endif /*HAVE_ROW_BASED_REPLICATION*/
 }
 
+
 static void fix_max_binlog_size(THD *thd, enum_var_type type)
 {
   DBUG_ENTER("fix_max_binlog_size");
diff --git a/sql/set_var.h b/sql/set_var.h
index 1049b154d47..5e264c68337 100644
--- a/sql/set_var.h
+++ b/sql/set_var.h
@@ -881,15 +881,20 @@ public:
   byte *value_ptr(THD *thd, enum_var_type type, LEX_STRING *base);
 };
 
+#ifdef HAVE_ROW_BASED_REPLICATION
 extern void fix_binlog_format_after_update(THD *thd, enum_var_type type);
+#endif
 
 class sys_var_thd_binlog_format :public sys_var_thd_enum
 {
 public:
   sys_var_thd_binlog_format(const char *name_arg, ulong SV::*offset_arg)
     :sys_var_thd_enum(name_arg, offset_arg,
-                      &binlog_format_typelib,
-                      fix_binlog_format_after_update)
+                      &binlog_format_typelib
+#ifdef HAVE_ROW_BASED_REPLICATION
+                      , fix_binlog_format_after_update
+#endif
+                      )
   {};
   bool is_readonly() const;
 };
diff --git a/sql/share/errmsg.txt b/sql/share/errmsg.txt
index 421da4d40b5..01fcfd1adf9 100644
--- a/sql/share/errmsg.txt
+++ b/sql/share/errmsg.txt
@@ -5835,3 +5835,5 @@ ER_EVENT_SET_VAR_ERROR
 ER_PARTITION_MERGE_ERROR
         eng "%s handler cannot be used in partitioned tables"
         swe "%s kan inte användas i en partitionerad tabell"
+ER_RBR_NOT_AVAILABLE
+        eng "The server was not built with row-based replication"
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 0a9e6472d9f..23603afc038 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -2134,7 +2134,9 @@ void THD::reset_sub_statement_state(Sub_statement_state *backup,
 
   if ((!lex->requires_prelocking() || is_update_query(lex->sql_command)) &&
       !current_stmt_binlog_row_based)
+  {
     options&= ~OPTION_BIN_LOG;
+  }    
   /* Disable result sets */
   client_capabilities &= ~CLIENT_MULTI_RESULTS;
   in_sub_stmt|= new_state;
@@ -2642,31 +2644,6 @@ int THD::binlog_flush_pending_rows_event(bool stmt_end)
 
     error= mysql_bin_log.flush_and_set_pending_rows_event(this, 0);
   }
-  else if (stmt_end && binlog_table_maps > 0)
-  {                      /* there is no pending event at this point */
-    /*
-      If pending is null and we are going to end the statement, we
-      have to write an extra, empty, binrow event so that the slave
-      knows to discard the tables it has received.  Otherwise, the
-      table maps written this far will be included in the table maps
-      for the following statement.
-
-      TODO: Remove the need for a dummy event altogether.  It can be
-      fixed if we can write table maps to a memory buffer before
-      writing the first binrow event.  We can then flush and clear the
-      memory buffer with table map events before writing the first
-      binrow event.  In the event of a crash, nothing is lost since
-      the table maps are only needed if there are binrow events.
-    */
-
-    Rows_log_event *ev=
-      new Write_rows_log_event(this, 0, ~0UL, 0, FALSE);
-    ev->set_flags(Rows_log_event::STMT_END_F);
-    binlog_set_pending_rows_event(ev);
-
-    error= mysql_bin_log.flush_and_set_pending_rows_event(this, 0);
-    binlog_table_maps= 0;
-  }
 
   DBUG_RETURN(error);
 }
@@ -2724,6 +2701,7 @@ int THD::binlog_query(THD::enum_binlog_query_type qtype,
       to how you treat this.
     */
   case THD::ROW_QUERY_TYPE:
+#ifdef HAVE_ROW_BASED_REPLICATION
     if (current_stmt_binlog_row_based)
     {
       /*
@@ -2744,6 +2722,7 @@ int THD::binlog_query(THD::enum_binlog_query_type qtype,
 #endif /*HAVE_ROW_BASED_REPLICATION*/
       DBUG_RETURN(0);
     }
+#endif
     /* Otherwise, we fall through */
   case THD::STMT_QUERY_TYPE:
     /*
@@ -2752,7 +2731,9 @@ int THD::binlog_query(THD::enum_binlog_query_type qtype,
      */
     {
       Query_log_event qinfo(this, query, query_len, is_trans, suppress_use);
+#ifdef HAVE_ROW_BASED_REPLICATION
       qinfo.flags|= LOG_EVENT_UPDATE_TABLE_MAP_VERSION_F;
+#endif
       /*
         Binlog table maps will be irrelevant after a Query_log_event
         (they are just removed on the slave side) so after the query
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 723dad715bd..b6283b6d174 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -908,8 +908,10 @@ public:
   /* container for handler's private per-connection data */
   void *ha_data[MAX_HA];
 
-#ifdef HAVE_ROW_BASED_REPLICATION
 #ifndef MYSQL_CLIENT
+  int binlog_setup_trx_data();
+
+#ifdef HAVE_ROW_BASED_REPLICATION
 
   /*
     Public interface to write RBR events to the binlog
@@ -939,7 +941,6 @@ public:
 				      RowsEventT* hint);
   Rows_log_event* binlog_get_pending_rows_event() const;
   void            binlog_set_pending_rows_event(Rows_log_event* ev);
-  int             binlog_setup_trx_data();
   
   my_size_t max_row_length_blob(TABLE* table, const byte *data) const;
   my_size_t max_row_length(TABLE* table, const byte *data) const
@@ -960,12 +961,15 @@ public:
 
 private:
   uint binlog_table_maps; // Number of table maps currently in the binlog
-
 public:
-
-#endif
+  uint get_binlog_table_maps() const {
+    return binlog_table_maps;
+  }
 #endif /* HAVE_ROW_BASED_REPLICATION */
+#endif /* MYSQL_CLIENT */
+
 #ifndef MYSQL_CLIENT
+public:
   enum enum_binlog_query_type {
       /*
         The query can be logged row-based or statement-based
@@ -1403,18 +1407,26 @@ public:
   inline void set_current_stmt_binlog_row_based_if_mixed()
   {
     if (variables.binlog_format == BINLOG_FORMAT_MIXED)
-      current_stmt_binlog_row_based= 1;
+      current_stmt_binlog_row_based= TRUE;
   }
   inline void set_current_stmt_binlog_row_based()
   {
-    current_stmt_binlog_row_based= 1;
+    current_stmt_binlog_row_based= TRUE;
+  }
+  inline void clear_current_stmt_binlog_row_based()
+  {
+    current_stmt_binlog_row_based= FALSE;
   }
+#endif
   inline void reset_current_stmt_binlog_row_based()
   {
-    current_stmt_binlog_row_based= test(variables.binlog_format ==
-                                        BINLOG_FORMAT_ROW);
+#ifdef HAVE_ROW_BASED_REPLICATION
+    current_stmt_binlog_row_based=
+      test(variables.binlog_format == BINLOG_FORMAT_ROW);
+#else
+    current_stmt_binlog_row_based= FALSE;
+#endif
   }
-#endif /*HAVE_ROW_BASED_REPLICATION*/
 };
 
 
@@ -1579,6 +1591,9 @@ class select_insert :public select_result_interceptor {
   bool send_eof();
   /* not implemented: select_insert is never re-used in prepared statements */
   void cleanup();
+
+protected:
+  MYSQL_LOCK *lock;
 };
 
 
@@ -1588,7 +1603,6 @@ class select_create: public select_insert {
   List<create_field> *extra_fields;
   List<Key> *keys;
   HA_CREATE_INFO *create_info;
-  MYSQL_LOCK *lock;
   Field **field;
 public:
   select_create (TABLE_LIST *table,
@@ -1598,10 +1612,10 @@ public:
 		 List<Item> &select_fields,enum_duplicates duplic, bool ignore)
     :select_insert (NULL, NULL, &select_fields, 0, 0, duplic, ignore),
     create_table(table), extra_fields(&fields_par),keys(&keys_par),
-    create_info(create_info_par),
-    lock(0)
+    create_info(create_info_par)
     {}
   int prepare(List<Item> &list, SELECT_LEX_UNIT *u);
+
   void binlog_show_create_table(TABLE **tables, uint count);
   void store_values(List<Item> &values);
   void send_error(uint errcode,const char *err);
diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index dd4748bc15c..73a0a9a69c2 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -42,8 +42,6 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
   ha_rows	deleted= 0;
   uint usable_index= MAX_KEY;
   SELECT_LEX   *select_lex= &thd->lex->select_lex;
-  bool          ha_delete_all_rows= 0;
-  ulonglong const saved_options= thd->options;
   DBUG_ENTER("mysql_delete");
 
   if (open_and_lock_tables(thd, table_list))
@@ -75,20 +73,19 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
     Test if the user wants to delete all rows and deletion doesn't have
     any side-effects (because of triggers), so we can use optimized
     handler::delete_all_rows() method.
+
+    If row-based replication is used, we also delete the table row by
+    row.
   */
   if (!using_limit && const_cond && (!conds || conds->val_int()) &&
       !(specialflag & (SPECIAL_NO_NEW_FUNC | SPECIAL_SAFE_MODE)) &&
-      !(table->triggers && table->triggers->has_delete_triggers()))
+      !(table->triggers && table->triggers->has_delete_triggers()) &&
+      !thd->current_stmt_binlog_row_based)
   {
     /* Update the table->file->stats.records number */
     table->file->info(HA_STATUS_VARIABLE | HA_STATUS_NO_LOCK);
     ha_rows const maybe_deleted= table->file->stats.records;
-    /*
-      If all rows shall be deleted, we (almost) always log this
-      statement-based (see [binlog], below), so we set this flag and
-      test it below.
-    */
-    ha_delete_all_rows= 1;
+    DBUG_PRINT("debug", ("Trying to use delete_all_rows()"));
     if (!(error=table->file->delete_all_rows()))
     {
       error= -1;				// ok
@@ -218,13 +215,6 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
   thd->proc_info="updating";
   will_batch= !table->file->start_bulk_delete();
 
-  /*
-    We saved the thread options above before clearing the
-    OPTION_BIN_LOG, and will restore below, effectively disabling the
-    binary log (unless it was already disabled, of course).
-  */
-  if (ha_delete_all_rows)
-    thd->options&= ~static_cast<ulonglong>(OPTION_BIN_LOG);
 
   table->mark_columns_needed_for_delete();
 
@@ -318,12 +308,6 @@ cleanup:
   delete select;
   transactional_table= table->file->has_transactions();
 
-  /*
-    Restore the saved value of the OPTION_BIN_LOG bit in the thread
-    options before executing binlog_query() below.
-  */
-  thd->options|= (saved_options & OPTION_BIN_LOG);
-
   /* See similar binlogging code in sql_update.cc, for comments */
   if ((error < 0) || (deleted && !transactional_table))
   {
@@ -338,11 +322,7 @@ cleanup:
         statement-based; otherwise, 'ha_delete_row()' was used to
         delete specific rows which we might log row-based.
       */
-      THD::enum_binlog_query_type const
-	query_type(ha_delete_all_rows && !table->file->is_injective() ?
-		   THD::STMT_QUERY_TYPE :
-		   THD::ROW_QUERY_TYPE);
-      int log_result= thd->binlog_query(query_type,
+      int log_result= thd->binlog_query(THD::ROW_QUERY_TYPE,
                                         thd->query, thd->query_length,
                                         transactional_table, FALSE);
 
@@ -1004,6 +984,9 @@ trunc_by_del:
   thd->options&= ~(ulong) (OPTION_BEGIN | OPTION_NOT_AUTOCOMMIT);
   ha_enable_transaction(thd, FALSE);
   mysql_init_select(thd->lex);
+#ifdef HAVE_ROW_BASED_REPLICATION
+  thd->clear_current_stmt_binlog_row_based();
+#endif
   error= mysql_delete(thd, table_list, (COND*) 0, (SQL_LIST*) 0,
                       HA_POS_ERROR, LL(0), TRUE);
   ha_enable_transaction(thd, TRUE);
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index f1f97400283..aea9aba41ec 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -2191,6 +2191,7 @@ select_insert::select_insert(TABLE_LIST *table_list_par, TABLE *table_par,
                              bool ignore_check_option_errors)
   :table_list(table_list_par), table(table_par), fields(fields_par),
    last_insert_id(0),
+   lock(0),
    insert_into_view(table_list_par && table_list_par->view != 0)
 {
   bzero((char*) &info,sizeof(info));
@@ -2378,7 +2379,36 @@ bool select_insert::send_data(List<Item> &values)
       DBUG_RETURN(1);
     }
   }
-  if (!(error= write_record(thd, table, &info)))
+
+  /*
+    The thd->lock lock contain the locks for the select part of the
+    statement and the 'lock' variable contain the write lock for the
+    currently locked table that is being created or inserted
+    into. However, the row-based replication will investigate the
+    thd->lock to decide what table maps are to be written, so this one
+    has to contain the tables locked for writing.  To be able to write
+    table map for the table being created, we temporarily set
+    THD::lock to select_insert::lock while writing the record to the
+    storage engine. We cannot set this elsewhere, since the execution
+    of a stored function inside the select expression might cause the
+    lock structures to be NULL.
+   */
+  
+  {
+    MYSQL_LOCK *saved_lock= NULL;
+    if (lock)
+    {
+      saved_lock= thd->lock;
+      thd->lock= lock;
+    }
+
+    error= write_record(thd, table, &info);
+    
+    if (lock)
+      thd->lock= saved_lock;
+  }
+
+  if (!error)
   {
     if (table->triggers || info.handle_duplicates == DUP_UPDATE)
     {
@@ -2723,15 +2753,34 @@ private:
 };
 
 
-int select_create::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
+int
+select_create::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
 {
-  MY_HOOKS hooks(this);
   DBUG_ENTER("select_create::prepare");
 
+  TABLEOP_HOOKS *hook_ptr= NULL;
+#ifdef HAVE_ROW_BASED_REPLICATION
+  class MY_HOOKS : public TABLEOP_HOOKS {
+  public:
+    MY_HOOKS(select_create *x) : ptr(x) { }
+    virtual void do_prelock(TABLE **tables, uint count)
+    {
+      if (ptr->get_thd()->current_stmt_binlog_row_based)
+        ptr->binlog_show_create_table(tables, count);
+    }
+
+  private:
+    select_create *ptr;
+  };
+
+  MY_HOOKS hooks(this);
+  hook_ptr= &hooks;
+#endif
+
   unit= u;
   if (!(table= create_table_from_items(thd, create_info, create_table,
                                        extra_fields, keys, &values, &lock,
-                                       &hooks)))
+                                       hook_ptr)))
     DBUG_RETURN(-1);				// abort() deletes table
 
   if (table->s->fields < values.elements)
@@ -2767,7 +2816,9 @@ int select_create::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
 }
 
 
-void select_create::binlog_show_create_table(TABLE **tables, uint count)
+#ifdef HAVE_ROW_BASED_REPLICATION
+void
+select_create::binlog_show_create_table(TABLE **tables, uint count)
 {
   /*
     Note 1: In RBR mode, we generate a CREATE TABLE statement for the
@@ -2786,9 +2837,7 @@ void select_create::binlog_show_create_table(TABLE **tables, uint count)
     schema that will do a close_thread_tables(), destroying the
     statement transaction cache.
   */
-#ifdef HAVE_ROW_BASED_REPLICATION
   DBUG_ASSERT(thd->current_stmt_binlog_row_based);
-#endif
   DBUG_ASSERT(tables && *tables && count > 0);
 
   char buf[2048];
@@ -2808,7 +2857,7 @@ void select_create::binlog_show_create_table(TABLE **tables, uint count)
                     /* is_trans */ TRUE,
                     /* suppress_use */ FALSE);
 }
-
+#endif // HAVE_ROW_BASED_REPLICATION
 
 void select_create::store_values(List<Item> &values)
 {