Merge mysql-5.5-bugteam -> mysql-5.5-innodb

13 files changed, 385 insertions, 199 deletions

diff --git a/sql/item_sum.cc b/sql/item_sum.cc
index 8cd8e1bb222..7aff7940c4d 100644
--- a/sql/item_sum.cc
+++ b/sql/item_sum.cc
@@ -1325,27 +1325,11 @@ void Item_sum_sum::fix_length_and_dec()
 bool Item_sum_sum::add()
 {
   DBUG_ENTER("Item_sum_sum::add");
-  bool arg_is_null;
   if (hybrid_type == DECIMAL_RESULT)
   {
-    my_decimal value, *val;
-    if (aggr->use_distinct_values)
-    {
-      /* 
-        We are aggregating distinct rows. Get the value from the distinct 
-        table pointer
-      */
-      Aggregator_distinct *daggr= (Aggregator_distinct *)aggr;
-      val= daggr->table->field[0]->val_decimal (&value);
-      arg_is_null= daggr->table->field[0]->is_null();
-    }
-    else
-    {
-      /* non-distinct aggregation */
-      val= args[0]->val_decimal(&value);
-      arg_is_null= args[0]->null_value;
-    }
-    if (!arg_is_null)
+    my_decimal value;
+    const my_decimal *val= aggr->arg_val_decimal(&value);
+    if (!aggr->arg_is_null())
     {
       my_decimal_add(E_DEC_FATAL_ERROR, dec_buffs + (curr_dec_buff^1),
                      val, dec_buffs + curr_dec_buff);
@@ -1355,25 +1339,8 @@ bool Item_sum_sum::add()
   }
   else
   {
-    double val;
-    if (aggr->use_distinct_values)
-    {
-      /* 
-        We are aggregating distinct rows. Get the value from the distinct 
-        table pointer
-      */
-      Aggregator_distinct *daggr= (Aggregator_distinct *)aggr;
-      val= daggr->table->field[0]->val_real ();
-      arg_is_null= daggr->table->field[0]->is_null();
-    }
-    else
-    {
-      /* non-distinct aggregation */
-      val= args[0]->val_real();
-      arg_is_null= args[0]->null_value;
-    }
-    sum+= val;
-    if (!arg_is_null)
+    sum+= aggr->arg_val_real();
+    if (!aggr->arg_is_null())
       null_value= 0;
   }
   DBUG_RETURN(0);
@@ -1471,6 +1438,45 @@ Aggregator_distinct::~Aggregator_distinct()
 }
 
 
+my_decimal *Aggregator_simple::arg_val_decimal(my_decimal *value)
+{
+  return item_sum->args[0]->val_decimal(value);
+}
+
+
+double Aggregator_simple::arg_val_real()
+{
+  return item_sum->args[0]->val_real();
+}
+
+
+bool Aggregator_simple::arg_is_null()
+{
+  return item_sum->args[0]->null_value;
+}
+
+
+my_decimal *Aggregator_distinct::arg_val_decimal(my_decimal * value)
+{
+  return use_distinct_values ? table->field[0]->val_decimal(value) :
+    item_sum->args[0]->val_decimal(value);
+}
+
+
+double Aggregator_distinct::arg_val_real()
+{
+  return use_distinct_values ? table->field[0]->val_real() :
+    item_sum->args[0]->val_real();
+}
+
+
+bool Aggregator_distinct::arg_is_null()
+{
+  return use_distinct_values ? table->field[0]->is_null() :
+    item_sum->args[0]->null_value;
+}
+
+
 Item *Item_sum_count::copy_or_same(THD* thd)
 {
   return new (thd->mem_root) Item_sum_count(thd, this);
@@ -1576,7 +1582,7 @@ bool Item_sum_avg::add()
 {
   if (Item_sum_sum::add())
     return TRUE;
-  if (!args[0]->null_value)
+  if (!aggr->arg_is_null())
     count++;
   return FALSE;
 }
@@ -2229,7 +2235,7 @@ void Item_sum_avg::reset_field()
 
 void Item_sum_bit::reset_field()
 {
-  reset();
+  reset_and_add();
   int8store(result_field->ptr, bits);
 }
 
diff --git a/sql/item_sum.h b/sql/item_sum.h
index 06a8c2c58a4..3ea79fb8cee 100644
--- a/sql/item_sum.h
+++ b/sql/item_sum.h
@@ -101,6 +101,15 @@ public:
   */
   virtual void endup() = 0;
 
+  /** Decimal value of being-aggregated argument */
+  virtual my_decimal *arg_val_decimal(my_decimal * value) = 0;
+  /** Floating point value of being-aggregated argument */
+  virtual double arg_val_real() = 0;
+  /**
+     NULLness of being-aggregated argument; can be called only after
+     arg_val_decimal() or arg_val_real().
+  */
+  virtual bool arg_is_null() = 0;
 };
 
 
@@ -304,6 +313,7 @@ class st_select_lex;
 class Item_sum :public Item_result_field
 {
   friend class Aggregator_distinct;
+  friend class Aggregator_simple;
 
 protected:
   /**
@@ -396,14 +406,22 @@ public:
   Item_sum(THD *thd, Item_sum *item);
   enum Type type() const { return SUM_FUNC_ITEM; }
   virtual enum Sumfunctype sum_func () const=0;
-  inline bool reset() { aggregator_clear(); return aggregator_add(); };
+  /**
+    Resets the aggregate value to its default and aggregates the current
+    value of its attribute(s).
+  */  
+  inline bool reset_and_add() 
+  { 
+    aggregator_clear(); 
+    return aggregator_add(); 
+  };
 
   /*
     Called when new group is started and results are being saved in
-    a temporary table. Similar to reset(), but must also store value in
-    result_field. Like reset() it is supposed to reset start value to
-    default.
-    This set of methods (reult_field(), reset_field, update_field()) of
+    a temporary table. Similarly to reset_and_add() it resets the 
+    value to its default and aggregates the value of its 
+    attribute(s), but must also store it in result_field. 
+    This set of methods (result_item(), reset_field, update_field()) of
     Item_sum is used only if quick_group is not null. Otherwise
     copy_or_same() is used to obtain a copy of this item.
   */
@@ -446,7 +464,7 @@ public:
       set_aggregator(with_distinct ?
                      Aggregator::DISTINCT_AGGREGATOR :
                      Aggregator::SIMPLE_AGGREGATOR);
-    reset();
+    aggregator_clear();
   }
   virtual void make_unique() { force_copy_fields= TRUE; }
   Item *get_tmp_table_item(THD *thd);
@@ -600,6 +618,9 @@ public:
   void clear(); 
   bool add();
   void endup();
+  virtual my_decimal *arg_val_decimal(my_decimal * value);
+  virtual double arg_val_real();
+  virtual bool arg_is_null();
 
   bool unique_walk_function(void *element);
   static int composite_key_cmp(void* arg, uchar* key1, uchar* key2);
@@ -623,6 +644,9 @@ public:
   void clear() { item_sum->clear(); }
   bool add() { return item_sum->add(); }
   void endup() {};
+  virtual my_decimal *arg_val_decimal(my_decimal * value);
+  virtual double arg_val_real();
+  virtual bool arg_is_null();
 };
 
 
diff --git a/sql/log.cc b/sql/log.cc
index 73e26ca2641..4873cfd49c3 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -209,11 +209,11 @@ class binlog_cache_data
 {
 public:
   binlog_cache_data(): m_pending(0), before_stmt_pos(MY_OFF_T_UNDEF),
-  incident(FALSE), changes_to_non_trans_temp_table_flag(FALSE)
-  {
-    cache_log.end_of_file= max_binlog_cache_size;
-  }
-
+  incident(FALSE), changes_to_non_trans_temp_table_flag(FALSE),
+  saved_max_binlog_cache_size(0), ptr_binlog_cache_use(0),
+  ptr_binlog_cache_disk_use(0)
+  { }
+  
   ~binlog_cache_data()
   {
     DBUG_ASSERT(empty());
@@ -257,11 +257,19 @@ public:
 
   void reset()
   {
+    compute_statistics();
     truncate(0);
     changes_to_non_trans_temp_table_flag= FALSE;
     incident= FALSE;
     before_stmt_pos= MY_OFF_T_UNDEF;
-    cache_log.end_of_file= max_binlog_cache_size;
+    /*
+      The truncate function calls reinit_io_cache that calls my_b_flush_io_cache
+      which may increase disk_writes. This breaks the disk_writes use by the
+      binary log which aims to compute the ratio between in-memory cache usage
+      and disk cache usage. To avoid this undesirable behavior, we reset the
+      variable after truncating the cache.
+    */
+    cache_log.disk_writes= 0;
     DBUG_ASSERT(empty());
   }
 
@@ -292,6 +300,36 @@ public:
       before_stmt_pos= MY_OFF_T_UNDEF;
   }
 
+  void set_binlog_cache_info(ulong param_max_binlog_cache_size,
+                             ulong *param_ptr_binlog_cache_use,
+                             ulong *param_ptr_binlog_cache_disk_use)
+  {
+    /*
+      The assertions guarantee that the set_binlog_cache_info is
+      called just once and information passed as parameters are
+      never zero.
+
+      This is done while calling the constructor binlog_cache_mngr.
+      We cannot set informaton in the constructor binlog_cache_data
+      because the space for binlog_cache_mngr is allocated through
+      a placement new.
+
+      In the future, we can refactor this and change it to avoid
+      the set_binlog_info. 
+    */
+    DBUG_ASSERT(saved_max_binlog_cache_size == 0 &&
+                param_max_binlog_cache_size != 0 &&
+                ptr_binlog_cache_use == 0 &&
+                param_ptr_binlog_cache_use != 0 &&
+                ptr_binlog_cache_disk_use == 0 &&
+                param_ptr_binlog_cache_disk_use != 0);
+
+    saved_max_binlog_cache_size= param_max_binlog_cache_size;
+    ptr_binlog_cache_use= param_ptr_binlog_cache_use;
+    ptr_binlog_cache_disk_use= param_ptr_binlog_cache_disk_use;
+    cache_log.end_of_file= saved_max_binlog_cache_size;
+  }
+
   /*
     Cache to store data before copying it to the binary log.
   */
@@ -321,6 +359,40 @@ private:
   */
   bool changes_to_non_trans_temp_table_flag;
 
+  /**
+    This function computes binlog cache and disk usage.
+  */
+  void compute_statistics()
+  {
+    if (!empty())
+    {
+      statistic_increment(*ptr_binlog_cache_use, &LOCK_status);
+      if (cache_log.disk_writes != 0)
+        statistic_increment(*ptr_binlog_cache_disk_use, &LOCK_status);
+    }
+  }
+
+  /*
+    Stores the values of maximum size of the cache allowed when this cache
+    is configured. This corresponds to either
+      . max_binlog_cache_size or max_binlog_stmt_cache_size.
+  */
+  ulong saved_max_binlog_cache_size;
+
+  /*
+    Stores a pointer to the status variable that keeps track of the in-memory 
+    cache usage. This corresponds to either
+      . binlog_cache_use or binlog_stmt_cache_use.
+  */
+  ulong *ptr_binlog_cache_use;
+
+  /*
+    Stores a pointer to the status variable that keeps track of the disk
+    cache usage. This corresponds to either
+      . binlog_cache_disk_use or binlog_stmt_cache_disk_use.
+  */
+  ulong *ptr_binlog_cache_disk_use;
+
   /*
     It truncates the cache to a certain position. This includes deleting the
     pending event.
@@ -334,7 +406,7 @@ private:
       set_pending(0);
     }
     reinit_io_cache(&cache_log, WRITE_CACHE, pos, 0, 0);
-    cache_log.end_of_file= max_binlog_cache_size;
+    cache_log.end_of_file= saved_max_binlog_cache_size;
   }
  
   binlog_cache_data& operator=(const binlog_cache_data& info);
@@ -343,7 +415,20 @@ private:
 
 class binlog_cache_mngr {
 public:
-  binlog_cache_mngr() {}
+  binlog_cache_mngr(ulong param_max_binlog_stmt_cache_size,
+                    ulong param_max_binlog_cache_size,
+                    ulong *param_ptr_binlog_stmt_cache_use,
+                    ulong *param_ptr_binlog_stmt_cache_disk_use,
+                    ulong *param_ptr_binlog_cache_use,
+                    ulong *param_ptr_binlog_cache_disk_use)
+  {
+     stmt_cache.set_binlog_cache_info(param_max_binlog_stmt_cache_size,
+                                      param_ptr_binlog_stmt_cache_use,
+                                      param_ptr_binlog_stmt_cache_disk_use);
+     trx_cache.set_binlog_cache_info(param_max_binlog_cache_size,
+                                     param_ptr_binlog_cache_use,
+                                     param_ptr_binlog_cache_disk_use);
+  }
 
   void reset_cache(binlog_cache_data* cache_data)
   {
@@ -1506,57 +1591,122 @@ static int binlog_close_connection(handlerton *hton, THD *thd)
 }
 
 /**
-  This function flushes a transactional cache upon commit/rollback.
+  This function flushes a cache upon commit/rollback.
 
-  @param thd        The thread whose transaction should be flushed
-  @param cache_mngr Pointer to the cache data to be flushed
-  @param end_ev     The end event either commit/rollback.
+  @param thd                The thread whose transaction should be flushed
+  @param cache_data         Pointer to the cache
+  @param end_ev             The end event either commit/rollback
+  @param is_transactional   The type of the cache: transactional or
+                            non-transactional
 
   @return
-    nonzero if an error pops up when flushing the transactional cache.
+    nonzero if an error pops up when flushing the cache.
 */
-static int
-binlog_flush_trx_cache(THD *thd, binlog_cache_mngr *cache_mngr,
-                       Log_event *end_ev)
+static inline int
+binlog_flush_cache(THD *thd, binlog_cache_data* cache_data, Log_event *end_evt,
+                   bool is_transactional)
 {
-  DBUG_ENTER("binlog_flush_trx_cache");
-  int error=0;
-  IO_CACHE *cache_log= &cache_mngr->trx_cache.cache_log;
-
-  /*
-    This function handles transactional changes and as such
-    this flag equals to true.
-  */
-  bool const is_transactional= TRUE;
-
-  if (thd->binlog_flush_pending_rows_event(TRUE, is_transactional))
-    DBUG_RETURN(1);
-  /*
-    Doing a commit or a rollback including non-transactional tables,
-    i.e., ending a transaction where we might write the transaction
-    cache to the binary log.
-
-    We can always end the statement when ending a transaction since
-    transactions are not allowed inside stored functions. If they
-    were, we would have to ensure that we're not ending a statement
-    inside a stored function.
-  */
-  error= mysql_bin_log.write(thd, &cache_mngr->trx_cache.cache_log, end_ev,
-                             cache_mngr->trx_cache.has_incident());
-  cache_mngr->reset_cache(&cache_mngr->trx_cache);
+  DBUG_ENTER("binlog_flush_cache");
+  int error= 0;
 
-  statistic_increment(binlog_cache_use, &LOCK_status);
-  if (cache_log->disk_writes != 0)
+  if (!cache_data->empty())
   {
-    statistic_increment(binlog_cache_disk_use, &LOCK_status);
-    cache_log->disk_writes= 0;
+    if (thd->binlog_flush_pending_rows_event(TRUE, is_transactional))
+      DBUG_RETURN(1);
+    /*
+      Doing a commit or a rollback including non-transactional tables,
+      i.e., ending a transaction where we might write the transaction
+      cache to the binary log.
+
+      We can always end the statement when ending a transaction since
+      transactions are not allowed inside stored functions. If they
+      were, we would have to ensure that we're not ending a statement
+      inside a stored function.
+    */
+    error= mysql_bin_log.write(thd, &cache_data->cache_log, end_evt,
+                               cache_data->has_incident());
   }
+  cache_data->reset();
 
-  DBUG_ASSERT(cache_mngr->trx_cache.empty());
+  DBUG_ASSERT(cache_data->empty());
   DBUG_RETURN(error);
 }
 
 /**
+  This function flushes the stmt-cache upon commit.
+
+  @param thd                The thread whose transaction should be flushed
+  @param cache_mngr         Pointer to the cache manager
+
+  @return
+    nonzero if an error pops up when flushing the cache.
+*/
+static inline int
+binlog_commit_flush_stmt_cache(THD *thd,
+                               binlog_cache_mngr *cache_mngr)
+{
+  Query_log_event end_evt(thd, STRING_WITH_LEN("COMMIT"),
+                          FALSE, FALSE, TRUE, 0);
+  return (binlog_flush_cache(thd, &cache_mngr->stmt_cache, &end_evt,
+                             FALSE));
+}
+
+/**
+  This function flushes the trx-cache upon commit.
+
+  @param thd                The thread whose transaction should be flushed
+  @param cache_mngr         Pointer to the cache manager
+
+  @return
+    nonzero if an error pops up when flushing the cache.
+*/
+static inline int
+binlog_commit_flush_trx_cache(THD *thd, binlog_cache_mngr *cache_mngr)
+{
+  Query_log_event end_evt(thd, STRING_WITH_LEN("COMMIT"),
+                          TRUE, FALSE, TRUE, 0);
+  return (binlog_flush_cache(thd, &cache_mngr->trx_cache, &end_evt,
+                             TRUE));
+}
+
+/**
+  This function flushes the trx-cache upon rollback.
+
+  @param thd                The thread whose transaction should be flushed
+  @param cache_mngr         Pointer to the cache manager
+
+  @return
+    nonzero if an error pops up when flushing the cache.
+*/
+static inline int
+binlog_rollback_flush_trx_cache(THD *thd, binlog_cache_mngr *cache_mngr)
+{
+  Query_log_event end_evt(thd, STRING_WITH_LEN("ROLLBACK"),
+                          TRUE, FALSE, TRUE, 0);
+  return (binlog_flush_cache(thd, &cache_mngr->trx_cache, &end_evt,
+                             TRUE));
+}
+
+/**
+  This function flushes the trx-cache upon commit.
+
+  @param thd                The thread whose transaction should be flushed
+  @param cache_mngr         Pointer to the cache manager
+  @param xid                Transaction Id
+
+  @return
+    nonzero if an error pops up when flushing the cache.
+*/
+static inline int
+binlog_commit_flush_trx_cache(THD *thd, binlog_cache_mngr *cache_mngr,
+                              my_xid xid)
+{
+  Xid_log_event end_evt(thd, xid);
+  return (binlog_flush_cache(thd, &cache_mngr->trx_cache, &end_evt,
+                             TRUE));
+}
+
+/**
   This function truncates the transactional cache upon committing or rolling
   back either a transaction or a statement.
 
@@ -1579,23 +1729,24 @@ binlog_truncate_trx_cache(THD *thd, binlog_cache_mngr *cache_mngr, bool all)
   */
   bool const is_transactional= TRUE;
 
-  DBUG_PRINT("info", ("thd->options={ %s%s}, transaction: %s",
+  DBUG_PRINT("info", ("thd->options={ %s %s}, transaction: %s",
                       FLAGSTR(thd->variables.option_bits, OPTION_NOT_AUTOCOMMIT),
                       FLAGSTR(thd->variables.option_bits, OPTION_BEGIN),
                       all ? "all" : "stmt"));
+
+  thd->binlog_remove_pending_rows_event(TRUE, is_transactional);
   /*
     If rolling back an entire transaction or a single statement not
     inside a transaction, we reset the transaction cache.
   */
-  thd->binlog_remove_pending_rows_event(TRUE, is_transactional);
   if (ending_trans(thd, all))
   {
     if (cache_mngr->trx_cache.has_incident())
       error= mysql_bin_log.write_incident(thd, TRUE);
 
-    cache_mngr->reset_cache(&cache_mngr->trx_cache);
-
     thd->clear_binlog_table_maps();
+
+    cache_mngr->reset_cache(&cache_mngr->trx_cache);
   }
   /*
     If rolling back a statement in a transaction, we truncate the
@@ -1620,51 +1771,6 @@ static int binlog_prepare(handlerton *hton, THD *thd, bool all)
 }
 
 /**
-  This function flushes the non-transactional to the binary log upon
-  committing or rolling back a statement.
-
-  @param thd        The thread whose transaction should be flushed
-  @param cache_mngr Pointer to the cache data to be flushed
-
-  @return
-    nonzero if an error pops up when flushing the non-transactional cache.
-*/
-static int
-binlog_flush_stmt_cache(THD *thd, binlog_cache_mngr *cache_mngr)
-{
-  int error= 0;
-  DBUG_ENTER("binlog_flush_stmt_cache");
-  /*
-    If we are flushing the statement cache, it means that the changes get
-    through otherwise the cache is empty and this routine should not be called.
-  */
-  DBUG_ASSERT(cache_mngr->stmt_cache.has_incident() == FALSE);
-  /*
-    This function handles non-transactional changes and as such this flag equals
-    to false.
-  */
-  bool const is_transactional= FALSE;
-  IO_CACHE *cache_log= &cache_mngr->stmt_cache.cache_log;
-
-  if (thd->binlog_flush_pending_rows_event(TRUE, is_transactional))
-    DBUG_RETURN(1);
-
-  Query_log_event qev(thd, STRING_WITH_LEN("COMMIT"), TRUE, FALSE, TRUE, 0);
-  if ((error= mysql_bin_log.write(thd, cache_log, &qev,
-                                  cache_mngr->stmt_cache.has_incident())))
-    DBUG_RETURN(error);
-  cache_mngr->reset_cache(&cache_mngr->stmt_cache);
-
-  statistic_increment(binlog_cache_use, &LOCK_status);
-  if (cache_log->disk_writes != 0)
-  {
-    statistic_increment(binlog_cache_disk_use, &LOCK_status);
-    cache_log->disk_writes= 0;
-  }
-  DBUG_RETURN(error);
-}
-
-/**
   This function is called once after each statement.
 
   It has the responsibility to flush the caches to the binary log on commits.
@@ -1692,7 +1798,7 @@ static int binlog_commit(handlerton *hton, THD *thd, bool all)
 
   if (!cache_mngr->stmt_cache.empty())
   {
-    binlog_flush_stmt_cache(thd, cache_mngr);
+    error= binlog_commit_flush_stmt_cache(thd, cache_mngr);
   }
 
   if (cache_mngr->trx_cache.empty())
@@ -1701,7 +1807,7 @@ static int binlog_commit(handlerton *hton, THD *thd, bool all)
       we're here because cache_log was flushed in MYSQL_BIN_LOG::log_xid()
     */
     cache_mngr->reset_cache(&cache_mngr->trx_cache);
-    DBUG_RETURN(0);
+    DBUG_RETURN(error);
   }
 
   /*
@@ -1710,17 +1816,15 @@ static int binlog_commit(handlerton *hton, THD *thd, bool all)
      - We are in a transaction and a full transaction is committed.
     Otherwise, we accumulate the changes.
   */
-  if (ending_trans(thd, all))
-  {
-    Query_log_event qev(thd, STRING_WITH_LEN("COMMIT"), TRUE, FALSE, TRUE, 0);
-    error= binlog_flush_trx_cache(thd, cache_mngr, &qev);
-  }
+  if (!error && ending_trans(thd, all))
+    error= binlog_commit_flush_trx_cache(thd, cache_mngr);
 
   /*
     This is part of the stmt rollback.
   */
   if (!all)
     cache_mngr->trx_cache.set_prev_position(MY_OFF_T_UNDEF);
+
   DBUG_RETURN(error);
 }
 
@@ -1737,7 +1841,7 @@ static int binlog_commit(handlerton *hton, THD *thd, bool all)
 static int binlog_rollback(handlerton *hton, THD *thd, bool all)
 {
   DBUG_ENTER("binlog_rollback");
-  int error=0;
+  int error= 0;
   binlog_cache_mngr *const cache_mngr=
     (binlog_cache_mngr*) thd_get_ha_data(thd, binlog_hton);
 
@@ -1757,16 +1861,16 @@ static int binlog_rollback(handlerton *hton, THD *thd, bool all)
   }
   else if (!cache_mngr->stmt_cache.empty())
   {
-    binlog_flush_stmt_cache(thd, cache_mngr);
+    error= binlog_commit_flush_stmt_cache(thd, cache_mngr);
   }
 
   if (cache_mngr->trx_cache.empty())
   {
-    /* 
+    /*
       we're here because cache_log was flushed in MYSQL_BIN_LOG::log_xid()
     */
     cache_mngr->reset_cache(&cache_mngr->trx_cache);
-    DBUG_RETURN(0);
+    DBUG_RETURN(error);
   }
 
   if (mysql_bin_log.check_write_error(thd))
@@ -1782,9 +1886,9 @@ static int binlog_rollback(handlerton *hton, THD *thd, bool all)
       We reach this point if the effect of a statement did not properly get into
       a cache and need to be rolled back.
     */
-    error= binlog_truncate_trx_cache(thd, cache_mngr, all);
+    error |= binlog_truncate_trx_cache(thd, cache_mngr, all);
   }
-  else
+  else if (!error)
   {  
     /*
       We flush the cache wrapped in a beging/rollback if:
@@ -1796,7 +1900,6 @@ static int binlog_rollback(handlerton *hton, THD *thd, bool all)
         . the format is MIXED, non-trans table was updated and
           aborting a single statement transaction;
     */
-
     if (ending_trans(thd, all) &&
         ((thd->variables.option_bits & OPTION_KEEP_LOG) ||
          (trans_has_updated_non_trans_table(thd) &&
@@ -1806,10 +1909,7 @@ static int binlog_rollback(handlerton *hton, THD *thd, bool all)
          (trans_has_updated_non_trans_table(thd) &&
           ending_single_stmt_trans(thd,all) &&
           thd->variables.binlog_format == BINLOG_FORMAT_MIXED)))
-    {
-      Query_log_event qev(thd, STRING_WITH_LEN("ROLLBACK"), TRUE, FALSE, TRUE, 0);
-      error= binlog_flush_trx_cache(thd, cache_mngr, &qev);
-    }
+      error= binlog_rollback_flush_trx_cache(thd, cache_mngr);
     /*
       Truncate the cache if:
         . aborting a single or multi-statement transaction or;
@@ -1832,11 +1932,12 @@ static int binlog_rollback(handlerton *hton, THD *thd, bool all)
     This is part of the stmt rollback.
   */
   if (!all)
-    cache_mngr->trx_cache.set_prev_position(MY_OFF_T_UNDEF); 
+    cache_mngr->trx_cache.set_prev_position(MY_OFF_T_UNDEF);
+
   DBUG_RETURN(error);
 }
 
-void MYSQL_BIN_LOG::set_write_error(THD *thd)
+void MYSQL_BIN_LOG::set_write_error(THD *thd, bool is_transactional)
 {
   DBUG_ENTER("MYSQL_BIN_LOG::set_write_error");
 
@@ -1846,9 +1947,20 @@ void MYSQL_BIN_LOG::set_write_error(THD *thd)
     DBUG_VOID_RETURN;
 
   if (my_errno == EFBIG)
-    my_message(ER_TRANS_CACHE_FULL, ER(ER_TRANS_CACHE_FULL), MYF(MY_WME));
+  {
+    if (is_transactional)
+    {
+      my_message(ER_TRANS_CACHE_FULL, ER(ER_TRANS_CACHE_FULL), MYF(MY_WME));
+    }
+    else
+    {
+      my_message(ER_STMT_CACHE_FULL, ER(ER_STMT_CACHE_FULL), MYF(MY_WME));
+    }
+  }
   else
+  {
     my_error(ER_ERROR_ON_WRITE, MYF(MY_WME), name, errno);
+  }
 
   DBUG_VOID_RETURN;
 }
@@ -1865,6 +1977,7 @@ bool MYSQL_BIN_LOG::check_write_error(THD *thd)
   switch (thd->stmt_da->sql_errno())
   {
     case ER_TRANS_CACHE_FULL:
+    case ER_STMT_CACHE_FULL:
     case ER_ERROR_ON_WRITE:
     case ER_BINLOG_LOGGING_IMPOSSIBLE:
       checked= TRUE;
@@ -4370,7 +4483,7 @@ int THD::binlog_setup_trx_data()
   cache_mngr= (binlog_cache_mngr*) my_malloc(sizeof(binlog_cache_mngr), MYF(MY_ZEROFILL));
   if (!cache_mngr ||
       open_cached_file(&cache_mngr->stmt_cache.cache_log, mysql_tmpdir,
-                       LOG_PREFIX, binlog_cache_size, MYF(MY_WME)) ||
+                       LOG_PREFIX, binlog_stmt_cache_size, MYF(MY_WME)) ||
       open_cached_file(&cache_mngr->trx_cache.cache_log, mysql_tmpdir,
                        LOG_PREFIX, binlog_cache_size, MYF(MY_WME)))
   {
@@ -4379,8 +4492,13 @@ int THD::binlog_setup_trx_data()
   }
   thd_set_ha_data(this, binlog_hton, cache_mngr);
 
-  cache_mngr= new (thd_get_ha_data(this, binlog_hton)) binlog_cache_mngr;
-
+  cache_mngr= new (thd_get_ha_data(this, binlog_hton))
+              binlog_cache_mngr(max_binlog_stmt_cache_size,
+                                max_binlog_cache_size,
+                                &binlog_stmt_cache_use,
+                                &binlog_stmt_cache_disk_use,
+                                &binlog_cache_use,
+                                &binlog_cache_disk_use);
   DBUG_RETURN(0);
 }
 
@@ -4632,7 +4750,7 @@ MYSQL_BIN_LOG::flush_and_set_pending_rows_event(THD *thd,
     */
     if (pending->write(file))
     {
-      set_write_error(thd);
+      set_write_error(thd, is_transactional);
       if (check_write_error(thd) && cache_data &&
           stmt_has_updated_non_trans_table(thd))
         cache_data->set_incident();
@@ -4657,6 +4775,7 @@ bool MYSQL_BIN_LOG::write(Log_event *event_info)
   bool error= 1;
   DBUG_ENTER("MYSQL_BIN_LOG::write(Log_event *)");
   binlog_cache_data *cache_data= 0;
+  bool is_trans_cache= FALSE;
 
   if (thd->binlog_evt_union.do_union)
   {
@@ -4717,7 +4836,7 @@ bool MYSQL_BIN_LOG::write(Log_event *event_info)
       binlog_cache_mngr *const cache_mngr=
         (binlog_cache_mngr*) thd_get_ha_data(thd, binlog_hton);
 
-      bool is_trans_cache= use_trans_cache(thd, event_info->use_trans_cache());
+      is_trans_cache= use_trans_cache(thd, event_info->use_trans_cache());
       file= cache_mngr->get_binlog_cache_log(is_trans_cache);
       cache_data= cache_mngr->get_binlog_cache_data(is_trans_cache);
 
@@ -4823,7 +4942,7 @@ unlock:
 
     if (error)
     {
-      set_write_error(thd);
+      set_write_error(thd, is_trans_cache);
       if (check_write_error(thd) && cache_data &&
           stmt_has_updated_non_trans_table(thd))
         cache_data->set_incident();
@@ -6322,15 +6441,14 @@ void TC_LOG_BINLOG::close()
 int TC_LOG_BINLOG::log_xid(THD *thd, my_xid xid)
 {
   DBUG_ENTER("TC_LOG_BINLOG::log");
-  Xid_log_event xle(thd, xid);
   binlog_cache_mngr *cache_mngr=
     (binlog_cache_mngr*) thd_get_ha_data(thd, binlog_hton);
   /*
     We always commit the entire transaction when writing an XID. Also
     note that the return value is inverted.
    */
-  DBUG_RETURN(!binlog_flush_stmt_cache(thd, cache_mngr) &&
-              !binlog_flush_trx_cache(thd, cache_mngr, &xle));
+  DBUG_RETURN(!binlog_commit_flush_stmt_cache(thd, cache_mngr) &&
+              !binlog_commit_flush_trx_cache(thd, cache_mngr, xid));
 }
 
 void TC_LOG_BINLOG::unlog(ulong cookie, my_xid xid)
diff --git a/sql/log.h b/sql/log.h
index d824d3afa26..4857f478cdf 100644
--- a/sql/log.h
+++ b/sql/log.h
@@ -415,7 +415,7 @@ public:
   bool write_incident(THD *thd, bool lock);
 
   int  write_cache(IO_CACHE *cache, bool lock_log, bool flush_and_sync);
-  void set_write_error(THD *thd);
+  void set_write_error(THD *thd, bool is_transactional);
   bool check_write_error(THD *thd);
 
   void start_union_events(THD *thd, query_id_t query_id_param);
diff --git a/sql/log_event.h b/sql/log_event.h
index 1841420ed86..a4d76e692a8 100644
--- a/sql/log_event.h
+++ b/sql/log_event.h
@@ -2506,7 +2506,7 @@ class Xid_log_event: public Log_event
    my_xid xid;
 
 #ifdef MYSQL_SERVER
-  Xid_log_event(THD* thd_arg, my_xid x): Log_event(thd_arg,0,0), xid(x) {}
+  Xid_log_event(THD* thd_arg, my_xid x): Log_event(thd_arg, 0, TRUE), xid(x) {}
 #ifdef HAVE_REPLICATION
   void pack_info(Protocol* protocol);
 #endif /* HAVE_REPLICATION */
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index cf8fd7f7d2e..49523a5b007 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -410,7 +410,8 @@ handlerton *heap_hton;
 handlerton *myisam_hton;
 handlerton *partition_hton;
 
-my_bool opt_readonly= 0, use_temp_pool, relay_log_purge;
+my_bool read_only= 0, opt_readonly= 0;
+my_bool use_temp_pool, relay_log_purge;
 my_bool relay_log_recovery;
 my_bool opt_sync_frm, opt_allow_suspicious_udfs;
 my_bool opt_secure_auth= 0;
@@ -461,6 +462,8 @@ ulonglong slave_type_conversions_options;
 ulong thread_cache_size=0;
 ulong binlog_cache_size=0;
 ulonglong  max_binlog_cache_size=0;
+ulong binlog_stmt_cache_size=0;
+ulonglong  max_binlog_stmt_cache_size=0;
 ulong query_cache_size=0;
 ulong refresh_version;  /* Increments on each reload */
 query_id_t global_query_id;
@@ -472,6 +475,7 @@ ulong delayed_insert_threads, delayed_insert_writes, delayed_rows_in_use;
 ulong delayed_insert_errors,flush_time;
 ulong specialflag=0;
 ulong binlog_cache_use= 0, binlog_cache_disk_use= 0;
+ulong binlog_stmt_cache_use= 0, binlog_stmt_cache_disk_use= 0;
 ulong max_connections, max_connect_errors;
 /**
   Limit of the total number of prepared statements in the server.
@@ -6407,6 +6411,8 @@ SHOW_VAR status_vars[]= {
   {"Aborted_connects",         (char*) &aborted_connects,       SHOW_LONG},
   {"Binlog_cache_disk_use",    (char*) &binlog_cache_disk_use,  SHOW_LONG},
   {"Binlog_cache_use",         (char*) &binlog_cache_use,       SHOW_LONG},
+  {"Binlog_stmt_cache_disk_use",(char*) &binlog_stmt_cache_disk_use,  SHOW_LONG},
+  {"Binlog_stmt_cache_use",    (char*) &binlog_stmt_cache_use,       SHOW_LONG},
   {"Bytes_received",           (char*) offsetof(STATUS_VAR, bytes_received), SHOW_LONGLONG_STATUS},
   {"Bytes_sent",               (char*) offsetof(STATUS_VAR, bytes_sent), SHOW_LONGLONG_STATUS},
   {"Com",                      (char*) com_status_vars, SHOW_ARRAY},
@@ -7344,6 +7350,8 @@ static int get_options(int *argc_ptr, char ***argv_ptr)
     test(global_system_variables.optimizer_switch &
          OPTIMIZER_SWITCH_ENGINE_CONDITION_PUSHDOWN);
 
+  opt_readonly= read_only;
+
   return 0;
 }
 
diff --git a/sql/mysqld.h b/sql/mysqld.h
index 376d8440619..5d8885ac277 100644
--- a/sql/mysqld.h
+++ b/sql/mysqld.h
@@ -107,7 +107,8 @@ extern my_bool opt_safe_show_db, opt_local_infile, opt_myisam_use_mmap;
 extern my_bool opt_slave_compressed_protocol, use_temp_pool;
 extern ulong slave_exec_mode_options;
 extern ulonglong slave_type_conversions_options;
-extern my_bool opt_readonly, lower_case_file_system;
+extern my_bool read_only, opt_readonly;
+extern my_bool lower_case_file_system;
 extern my_bool opt_enable_named_pipe, opt_sync_frm, opt_allow_suspicious_udfs;
 extern my_bool opt_secure_auth;
 extern char* opt_secure_file_priv;
@@ -152,6 +153,7 @@ extern ulonglong keybuff_size;
 extern ulonglong thd_startup_options;
 extern ulong thread_id;
 extern ulong binlog_cache_use, binlog_cache_disk_use;
+extern ulong binlog_stmt_cache_use, binlog_stmt_cache_disk_use;
 extern ulong aborted_threads,aborted_connects;
 extern ulong delayed_insert_timeout;
 extern ulong delayed_insert_limit, delayed_queue_size;
@@ -171,8 +173,9 @@ extern uint  slave_net_timeout;
 extern uint max_user_connections;
 extern ulong what_to_log,flush_time;
 extern ulong max_prepared_stmt_count, prepared_stmt_count;
-extern ulong binlog_cache_size, open_files_limit;
-extern ulonglong max_binlog_cache_size;
+extern ulong open_files_limit;
+extern ulong binlog_cache_size, binlog_stmt_cache_size;
+extern ulonglong max_binlog_cache_size, max_binlog_stmt_cache_size;
 extern ulong max_binlog_size, max_relay_log_size;
 extern ulong opt_binlog_rows_event_max_size;
 extern ulong rpl_recovery_rank, thread_cache_size;
diff --git a/sql/opt_range.cc b/sql/opt_range.cc
index 2ac860d25e3..f39037e9db9 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -11391,7 +11391,7 @@ void QUICK_GROUP_MIN_MAX_SELECT::update_min_result()
 
   min_functions_it->rewind();
   while ((min_func= (*min_functions_it)++))
-    min_func->reset();
+    min_func->reset_and_add();
 }
 
 
@@ -11423,7 +11423,7 @@ void QUICK_GROUP_MIN_MAX_SELECT::update_max_result()
 
   max_functions_it->rewind();
   while ((max_func= (*max_functions_it)++))
-    max_func->reset();
+    max_func->reset_and_add();
 }
 
 
diff --git a/sql/opt_sum.cc b/sql/opt_sum.cc
index e020c94a3bd..a2f8b9c4447 100644
--- a/sql/opt_sum.cc
+++ b/sql/opt_sum.cc
@@ -424,7 +424,7 @@ int opt_sum_query(TABLE_LIST *tables, List<Item> &all_fields,COND *conds)
           item_sum->aggregator_clear();
         }
         else
-          item_sum->reset();
+          item_sum->reset_and_add();
         item_sum->make_const();
         recalc_const_item= 1;
         break;
diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index be97afe055a..b0dc4d9195b 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/sql/share/errmsg-utf8.txt
@@ -6392,3 +6392,5 @@ ER_SLAVE_HEARTBEAT_VALUE_OUT_OF_RANGE_MIN
 ER_SLAVE_HEARTBEAT_VALUE_OUT_OF_RANGE_MAX
   eng "The requested value for the heartbeat period exceeds the value of `slave_net_timeout' seconds. A sensible value for the period should be less than the timeout."
 
+ER_STMT_CACHE_FULL  
+        eng "Multi-row statements required more than 'max_binlog_stmt_cache_size' bytes of storage; increase this mysqld variable and try again"
diff --git a/sql/sql_db.cc b/sql/sql_db.cc
index 292fbba2352..38f39ec0be7 100644
--- a/sql/sql_db.cc
+++ b/sql/sql_db.cc
@@ -794,14 +794,9 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
     }
   }
 
-  thd->push_internal_handler(&err_handler);
-
   if (find_db_tables_and_rm_known_files(thd, dirp, db, path, &tables,
                                         &found_other_files))
-  {
-    thd->pop_internal_handler();
     goto exit;
-  }
 
   /*
     Disable drop of enabled log tables, must be done before name locking.
@@ -815,7 +810,6 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
                              table->table_name_length, table->table_name, true))
       {
         my_error(ER_BAD_LOG_STATEMENT, MYF(0), "DROP");
-        thd->pop_internal_handler();
         goto exit;
       }
     }
@@ -825,10 +819,7 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
   if (lock_table_names(thd, tables, NULL, thd->variables.lock_wait_timeout,
                        MYSQL_OPEN_SKIP_TEMPORARY) ||
       lock_db_routines(thd, db))
-  {
-    thd->pop_internal_handler();
     goto exit;
-  }
 
   /* mysql_ha_rm_tables() requires a non-null TABLE_LIST. */
   if (tables)
@@ -841,6 +832,7 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
     deleted_tables++;
   }
 
+  thd->push_internal_handler(&err_handler);
   if (thd->killed ||
       (tables && mysql_rm_table_no_locks(thd, tables, true, false, true, true)))
   {
@@ -1094,8 +1086,16 @@ static bool find_db_tables_and_rm_known_files(THD *thd, MY_DIR *dirp,
     else
     {
       strxmov(filePath, path, "/", file->name, NullS);
-      if (mysql_file_delete_with_symlink(key_file_misc, filePath, MYF(MY_WME)))
-	DBUG_RETURN(true);
+      /*
+        We ignore ENOENT error in order to skip files that was deleted
+        by concurrently running statement like REAPIR TABLE ...
+      */
+      if (my_delete_with_symlink(filePath, MYF(0)) &&
+          my_errno != ENOENT)
+      {
+        my_error(EE_DELETE, MYF(0), filePath, my_errno);
+        DBUG_RETURN(true);
+      }
     }
   }
   *tables= tot_list;
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index ef2dd1d76e1..e4772bf3fb2 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -15850,7 +15850,7 @@ init_sum_functions(Item_sum **func_ptr, Item_sum **end_ptr)
 {
   for (; func_ptr != end_ptr ;func_ptr++)
   {
-    if ((*func_ptr)->reset())
+    if ((*func_ptr)->reset_and_add())
       return 1;
   }
   /* If rollup, calculate the upper sum levels */
diff --git a/sql/sys_vars.cc b/sql/sys_vars.cc
index 5c9df82ddac..68bb77d467f 100644
--- a/sql/sys_vars.cc
+++ b/sql/sys_vars.cc
@@ -236,14 +236,23 @@ static Sys_var_charptr Sys_basedir(
        IN_FS_CHARSET, DEFAULT(0));
 
 static Sys_var_ulong Sys_binlog_cache_size(
-       "binlog_cache_size", "The size of the cache to "
-       "hold the SQL statements for the binary log during a "
-       "transaction. If you often use big, multi-statement "
-       "transactions you can increase this to get more performance",
+       "binlog_cache_size", "The size of the transactional cache for "
+       "updates to transactional engines for the binary log. "
+       "If you often use transactions containing many statements, "
+       "you can increase this to get more performance",
        GLOBAL_VAR(binlog_cache_size),
        CMD_LINE(REQUIRED_ARG),
        VALID_RANGE(IO_SIZE, ULONG_MAX), DEFAULT(32768), BLOCK_SIZE(IO_SIZE));
 
+static Sys_var_ulong Sys_binlog_stmt_cache_size(
+       "binlog_stmt_cache_size", "The size of the statement cache for "
+       "updates to non-transactional engines for the binary log. "
+       "If you often use statements updating a great number of rows, "
+       "you can increase this to get more performance",
+       GLOBAL_VAR(binlog_stmt_cache_size),
+       CMD_LINE(REQUIRED_ARG),
+       VALID_RANGE(IO_SIZE, ULONG_MAX), DEFAULT(32768), BLOCK_SIZE(IO_SIZE));
+
 static bool check_has_super(sys_var *self, THD *thd, set_var *var)
 {
   DBUG_ASSERT(self->scope() != sys_var::GLOBAL);// don't abuse check_has_super()
@@ -1031,13 +1040,20 @@ static Sys_var_ulong Sys_max_allowed_packet(
 
 static Sys_var_ulonglong Sys_max_binlog_cache_size(
        "max_binlog_cache_size",
-       "Can be used to restrict the total size used to cache a "
-       "multi-transaction query",
+       "Sets the total size of the transactional cache",
        GLOBAL_VAR(max_binlog_cache_size), CMD_LINE(REQUIRED_ARG),
        VALID_RANGE(IO_SIZE, ULONGLONG_MAX),
        DEFAULT((ULONGLONG_MAX/IO_SIZE)*IO_SIZE),
        BLOCK_SIZE(IO_SIZE));
 
+static Sys_var_ulonglong Sys_max_binlog_stmt_cache_size(
+       "max_binlog_stmt_cache_size",
+       "Sets the total size of the statement cache",
+       GLOBAL_VAR(max_binlog_stmt_cache_size), CMD_LINE(REQUIRED_ARG),
+       VALID_RANGE(IO_SIZE, ULONGLONG_MAX),
+       DEFAULT((ULONGLONG_MAX/IO_SIZE)*IO_SIZE),
+       BLOCK_SIZE(IO_SIZE));
+
 static bool fix_max_binlog_size(sys_var *self, THD *thd, enum_var_type type)
 {
   mysql_bin_log.set_max_size(max_binlog_size);
@@ -1436,7 +1452,6 @@ static Sys_var_ulong Sys_read_buff_size(
        VALID_RANGE(IO_SIZE*2, INT_MAX32), DEFAULT(128*1024),
        BLOCK_SIZE(IO_SIZE));
 
-static my_bool read_only;
 static bool check_read_only(sys_var *self, THD *thd, set_var *var)
 {
   /* Prevent self dead-lock */
@@ -1520,6 +1535,16 @@ static bool fix_read_only(sys_var *self, THD *thd, enum_var_type type)
   read_only= opt_readonly;
   DBUG_RETURN(result);
 }
+
+
+/**
+  The read_only boolean is always equal to the opt_readonly boolean except
+  during fix_read_only(); when that function is entered, opt_readonly is
+  the pre-update value and read_only is the post-update value.
+  fix_read_only() compares them and runs needed operations for the
+  transition (especially when transitioning from false to true) and
+  synchronizes both booleans in the end.
+*/
 static Sys_var_mybool Sys_readonly(
        "read_only",
        "Make all non-temporary tables read-only, with the exception for "