bk commit into 5.1 tree (mats:1.1804)

[Mats Kindahl]

Below is the list of changes that have just been committed into a local
5.1 repository of mats. When mats does a push these changes will
be propagated to the main repository and, within 24 hours after the
push, to the public repository.
For information on how to access the public repository
see http://dev.mysql.com/doc/mysql/en/installing-source-tree.html

ChangeSet
  1.1804 05/04/22 20:28:53 mats@stripped +9 -0
  WL#1012: Several fixes to handle transactions and binary/relay log
  positions correctly.  Also added more efficient handling of tables
  when applying events at slave side.

  sql/sql_update.cc
    1.152 05/04/22 20:28:44 mats@stripped +0 -4
    Removed extreneous calls introduced earlier.

  sql/sql_insert.cc
    1.146 05/04/22 20:28:44 mats@stripped +0 -4
    Removed extreneous calls introduced earlier.

  sql/sql_delete.cc
    1.144 05/04/22 20:28:44 mats@stripped +0 -4
    Removed extreneous calls introduced earlier.

  sql/sql_class.h
    1.230 05/04/22 20:28:44 mats@stripped +25 -8
    Removing redundant member functions from THD.
    Adding member functions to THD structure to allow notification of pending
commit/rollback.

  sql/sql_class.cc
    1.175 05/04/22 20:28:43 mats@stripped +23 -9
    Flushing pending events at end of statement.
    Writing table map if row-request for unmapped table arrives.

  sql/log_event.cc
    1.173 05/04/22 20:28:43 mats@stripped +80 -48
    Added calls to handle table filtering.
    Using lock/unlock of tables on a per-event basis.
    Fixed error that did not step group_relay_log_pos properly.

  sql/log.cc
    1.160 05/04/22 20:28:43 mats@stripped +1 -1
    Removed extreneous call.

  sql/handler.h
    1.134 05/04/22 20:28:43 mats@stripped +0 -3
    Removed redundant member functions.

  sql/handler.cc
    1.152 05/04/22 20:28:42 mats@stripped +2 -20
    Added calls to member functions to let the THD prepare for commit/rollback.
    Removed redundant member functions.

# This is a BitKeeper patch.  What follows are the unified diffs for the
# set of deltas contained in the patch.  The rest of the patch, the part
# that BitKeeper cares about, is below these diffs.
# User:	mats
# Host:	romeo.kindahl.net
# Root:	/home/bk/w1012-mysql-5.1

--- 1.151/sql/handler.cc	2005-04-20 13:27:11 +02:00
+++ 1.152/sql/handler.cc	2005-04-22 20:28:42 +02:00
@@ -574,6 +574,7 @@
   my_xid xid= thd->transaction.xid.get_my_xid();
   DBUG_ENTER("ha_commit_trans");
 #ifdef USING_TRANSACTIONS
+  thd->prepare_for_commit(all);
   if (trans->nht)
   {
     DBUG_EXECUTE_IF("crash_commit_before", abort(););
@@ -669,6 +670,7 @@
   bool is_real_trans=all || thd->transaction.all.nht == 0;
   DBUG_ENTER("ha_rollback_trans");
 #ifdef USING_TRANSACTIONS
+  thd->prepare_for_rollback(all);
   if (trans->nht)
   {
     for (handlerton **ht=trans->ht; *ht; ht++)
@@ -2481,23 +2483,3 @@
   DBUG_PRINT("exit", ("error = %d", 0));
   DBUG_RETURN(0);
 }    
-  
-int handler::
-ha_stmt_begin() 
-{ 
-#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
-  return current_thd->transaction_begin(); 
-#else
-  return 0;
-#endif
-}
-
-int handler::
-ha_stmt_end()
-{ 
-#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
-  return current_thd->transaction_end(); 
-#else
-  return 0;
-#endif
-}

--- 1.133/sql/handler.h	2005-04-08 11:54:30 +02:00
+++ 1.134/sql/handler.h	2005-04-22 20:28:43 +02:00
@@ -557,9 +557,6 @@
   int ha_update_row(const byte *old_data, byte *new_data);
   int ha_delete_row(const byte *buf);
   
-  int ha_stmt_begin();
-  int ha_stmt_end();
-
   virtual int index_read(byte * buf, const byte * key,
 			 uint key_len, enum ha_rkey_function find_flag)
    { return  HA_ERR_WRONG_COMMAND; }

--- 1.159/sql/log.cc	2005-04-15 11:35:38 +02:00
+++ 1.160/sql/log.cc	2005-04-22 20:28:43 +02:00
@@ -1601,7 +1601,7 @@
         mixed trans/non-trans table types the best possible in binlogging)
       - or if the event asks for it (cache_stmt == true).
     */
-    if (opt_using_transactions && thd && event_info->can_be_cached())
+    if (opt_using_transactions && thd)
     {
       IO_CACHE *trans_log= (IO_CACHE*)thd->ha_data[binlog_hton.slot];
 

--- 1.172/sql/log_event.cc	2005-04-21 07:34:04 +02:00
+++ 1.173/sql/log_event.cc	2005-04-22 20:28:43 +02:00
@@ -4915,22 +4915,37 @@
   THD* const thd = current_thd;
   char const* row_start = m_rows_buf;
 
-  do_before_row_operations(table);
-  while (error == 0 && row_start < m_rows_end) {
-    DBUG_PRINT("info", ("row_start = %p, m_rows_end = %p", 
-			row_start, m_rows_end));
-    char const* row_end = do_prepare_row(thd, table, row_start);
-    DBUG_PRINT("info", ("row_start = %p, row_end = %p, (length %u)", 
-			row_start, row_end, row_end - row_start));
-    DBUG_ASSERT(row_end != NULL);
-    DBUG_ASSERT(row_end <= m_rows_end);
-    error = do_exec_row(table, rli);
-    DBUG_PRINT("info", ("error = %d", error));
-    row_start = row_end;
+  if (table) 
+  {	      // table == NULL means that this table should not be replicated.
+    TABLE_LIST table_list;
+    memset(&table_list, 0, sizeof(table_list));
+    table_list.db = const_cast<char*>(table->s->db);
+    table_list.alias = table_list.table_name 
+      = const_cast<char*>(table->s->table_name);
+    table_list.lock_type = TL_WRITE;
+    table_list.next_global = table_list.next_local = 0;
+    table_list.table= table;
+
+    lock_tables(thd, &table_list, 1 /* count */);
+
+    do_before_row_operations(table);
+    while (error == 0 && row_start < m_rows_end) {
+      DBUG_PRINT("info", ("row_start = %p, m_rows_end = %p", 
+			  row_start, m_rows_end));
+      char const* row_end = do_prepare_row(thd, table, row_start);
+      DBUG_PRINT("info", ("row_start = %p, row_end = %p, (length %u)", 
+			  row_start, row_end, row_end - row_start));
+      DBUG_ASSERT(row_end != NULL);
+      DBUG_ASSERT(row_end <= m_rows_end);
+      error = do_exec_row(table, rli);
+      DBUG_PRINT("info", ("error = %d", error));
+      row_start = row_end;
+    }
+    do_after_row_operations(table);
   }
-  do_after_row_operations(table);
 
-  if (m_flags & TRANS_END_F) {
+  if (m_flags & TRANS_END_F) 
+  {
     // This is the end of a statement or transaction, so close (and unlock)
     // the tables we opened when processing the Table_map_log_event starting
     // the statement.
@@ -4946,18 +4961,23 @@
     ha_commit(thd);
     close_thread_tables(thd);
   }
+  else if (thd->lock)
+  {
+    // if transactional...
+    error= ha_autocommit_or_rollback(thd,error);
+    mysql_unlock_tables(thd, thd->lock);
+    thd->lock=0;
+  }
     
   /*
     This is copied from the Log_event::exec_event(). We step the relay log
     positions ourself, since the generic version does not behave the way we
     want.
    */
-  DBUG_PRINT("info", ("TRANS_END_F is %s", 
-		      (m_flags & TRANS_END_F) ? "set" : "clear"));
-  DBUG_PRINT("info", ("OPTION_BEGIN is %s", 
-		      (thd->options & OPTION_BEGIN) ? "set" : "clear"));
-  if ((thd->options & OPTION_BEGIN) 
-      || !(m_flags & TRANS_END_F))
+  DBUG_PRINT("info", ("%sTRANS_END_F && %sOPTION_BEGIN", 
+		      ((m_flags & TRANS_END_F) ? "" : "!"),
+		      ((thd->options & OPTION_BEGIN) ? "" : "!")));
+  if ((thd->options & OPTION_BEGIN) || !(m_flags & TRANS_END_F))
   {
     DBUG_PRINT("info", ("calling inc_event_relay_log_pos()"));
     rli->inc_event_relay_log_pos();
@@ -4966,7 +4986,7 @@
   {
     DBUG_PRINT("info", ("calling inc_group_relay_log_pos() "
 			"and flush_relay_log_info()"));
-    rli->inc_group_relay_log_pos(0);
+    rli->inc_group_relay_log_pos(log_pos);
     flush_relay_log_info(rli);
     rli->last_master_timestamp= when;
   }
@@ -5170,7 +5190,7 @@
     open_tables() and lock_tables().
 */
 static unsigned int
-find_tables(THD* thd, TABLE_LIST* table_list, unsigned int *count)
+find_tables(THD* thd, TABLE_LIST* table_list, unsigned int count)
 {
   DBUG_ENTER("find_tables(THD*, TABLE_LIST*, unsigned int*)");
 
@@ -5193,9 +5213,6 @@
     }
   }
 
-  if (count)
-    *count = result;
-
   DBUG_PRINT("return", ("result=%d", result));
   DBUG_RETURN(result);
 }
@@ -5212,46 +5229,52 @@
   thd->query_id = next_query_id();
   pthread_mutex_unlock(&LOCK_thread_count);
 
-  /*
-    Open the table if it is not already open and lock it afterwards.
-  */
   TABLE_LIST table_list;
   memset(&table_list, 0, sizeof(table_list));
   table_list.db = const_cast<char*>(m_dbnam);
   table_list.alias = table_list.table_name = const_cast<char*>(m_tblnam);
   table_list.lock_type = TL_WRITE;
   table_list.next_global = table_list.next_local = 0;
+  table_list.updating = 1;
 
-  unsigned int count = 1;
-  TABLE_LIST* tables = &table_list;
-  if (find_tables(thd, &table_list, NULL) == 0) 
-  {
-    open_tables(thd, &tables, &count);
-  } 
-  lock_tables(thd, &table_list, count);
-  DBUG_PRINT("info", ("thd->locked_tables=%p, thd->lock=%p",   
-		      thd->locked_tables, thd->lock));
+  int error = 0;
 
-  /*
-    Now we have opened and locked the table. Now we can change the state of
-    the instance and update the values.
-  */
-  m_table = table_list.table;
+  if (!table_rules_on || tables_ok(thd, &table_list))
+  {
+    /*
+      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.
+    */
+    unsigned int count = 1;
+    if (find_tables(thd, &table_list, count) == 0) 
+    {
+      TABLE_LIST* tables = &table_list;
+      open_tables(thd, &tables, &count);
+    } 
+
+    /*
+      Now we have opened and locked the table. Now we can change the state of
+      this instance and update the values.
+    */
+    m_table = table_list.table;
 
-  int error = rli->m_table_map.set_table(m_table_id, m_table);
+    error = rli->m_table_map.set_table(m_table_id, m_table);
+  }
 
   /* 
      We explicitly do not call Log_event::exec_event() here since we do not
      want the relay log position to be flushed to disk. The flushing will be
      done by the last Rows_log_event that either ends a statement (outside a
      transaction) or a transaction.
-
+     
      A table map event can *never* end a transaction or a statement, so we
      just step the relay log position.
-   */
+  */
 
   rli->inc_event_relay_log_pos();
-
+  
   DBUG_RETURN(error);
 }
 #endif // HAVE_REPLICATION
@@ -5281,12 +5304,13 @@
   DBUG_PRINT("info", ("m_tbllen = %d; m_tblnam = %s", 
 		      m_tbllen, m_tblnam));
 
+  DBUG_ASSERT(m_dbnam != NULL);
+  DBUG_ASSERT(m_tblnam != NULL);
+
   // To ensure that we can use VLE in the future.
   DBUG_ASSERT(m_dblen < 128);
   DBUG_ASSERT(m_tbllen < 128);
   DBUG_ASSERT(m_colcnt < 128);
-  DBUG_ASSERT(m_dbnam != NULL);
-  DBUG_ASSERT(m_tblnam != NULL);
 
   byte const dbuf[] = { m_dblen };
   byte const tbuf[] = { m_tbllen };
@@ -5363,12 +5387,20 @@
 int Write_rows_log_event::
 do_before_row_operations(TABLE* table)
 {
+  // idempotency handling
+  thd->lex->sql_command= SQLCOM_REPLACE; // needed for innodb and ndbcluster
+  table->file->extra(HA_EXTRA_IGNORE_DUP_KEY); // needed for ndbcluster
+  // TODO: Ensure that myisam allow write_row() with duplicate key.
+ 
+  // TODO: better if we know how many...
+  table->file->start_bulk_insert(0);
   return 0;
 }
 
 void Write_rows_log_event::
 do_after_row_operations(TABLE* table)
 {
+  table->file->end_bulk_insert();
 }
 
 char const* Write_rows_log_event::

--- 1.174/sql/sql_class.cc	2005-04-20 13:27:11 +02:00
+++ 1.175/sql/sql_class.cc	2005-04-22 20:28:43 +02:00
@@ -1595,6 +1595,18 @@
   /* Note that free_list is freed in cleanup_after_query() */
 
   /*
+    Flush pending event if we are *not* in a transaction.  This will only do
+    something if we are processing a non-transactional table in statement
+    mode.  Otherwise, the pending event will already have been flushed.
+   */
+#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+  if (!(options & OPTION_BEGIN))
+  {
+    flush_pending_event(true);
+  }
+#endif
+
+  /*
     Don't free mem_root, as mem_root is freed in the end of dispatch_command
     (once for any command).
   */
@@ -1864,25 +1876,28 @@
 
   if (pending) 
   {
-      DBUG_PRINT("info", ("pending=%p [type_code=%d, size=%d]",
-			  pending, 
-			  pending->get_type_code(),
-			  pending->get_data_size()));
+    DBUG_PRINT("info", ("pending=%p [type_code=%d, size=%d]",
+			pending, 
+			pending->get_type_code(),
+			pending->get_data_size()));
   }
   else 
   {
-      DBUG_PRINT("info", ("pending=%p", pending));
+    DBUG_PRINT("info", ("pending=%p", pending));
   }
 
   DBUG_PRINT("info", ("opt_binlog_rows_event_max_size=%d", 
 		      opt_binlog_rows_event_max_size));
 
-  // Check if the current event is non-NULL and a write-rows event. 
+  // Check if the current event is non-NULL and a write-rows event. Also check
+  // if the table provided is mapped: if it is not, then we have switched to
+  // writing to a new table.
   if (!pending
       || pending->server_id != server_id
       || pending->get_type_code() != type_code
       || pending->get_data_size() + needed > opt_binlog_rows_event_max_size
-      || pending->get_cols() != cols) 
+      || pending->get_cols() != cols
+      || !mysql_bin_log.is_table_mapped(table)) 
   {
     // If not, flush the event and create a new RowsEventT.
     
@@ -1891,8 +1906,7 @@
       write_table_map(table);
     ulong const tid = mysql_bin_log.get_table_id(table);
 
-    Rows_log_event* const 
-	ev = new RowsEventT(this, table, tid, cols);
+    Rows_log_event* const ev = new RowsEventT(this, table, tid, cols);
     ev->server_id = server_id; // I don't like this, it's too easy to forget. 
     if (flush_and_set_pending_event(ev))
       DBUG_RETURN(NULL);

--- 1.229/sql/sql_class.h	2005-04-20 13:27:11 +02:00
+++ 1.230/sql/sql_class.h	2005-04-22 20:28:44 +02:00
@@ -1113,14 +1113,6 @@
 
   void set_server_id(uint32 sid) { server_id = sid; }
 
-  int transaction_begin() { 
-    return 0;
-  }
-
-  int transaction_end() { 
-    return flush_pending_event(true); 
-  }
-
 private:
   /*
     Member functions to handle pending event for row-level logging.
@@ -1144,6 +1136,31 @@
 #endif
 
 public:
+
+  /*
+    This function will be called from ha_commit_trans() to prepare for a
+    commit.
+  */
+  int prepare_for_commit(bool all) { 
+#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+    return flush_pending_event(true); 
+#else
+    return 0;
+#endif
+  }
+
+  /*
+    This function will be called from ha_rollback_trans() to prepare for a
+    rollback.
+  */
+  int prepare_for_rollback(bool all) { 
+#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+    return flush_pending_event(true); 
+#else
+    return 0;
+#endif
+  }
+
   struct st_transactions {
     SAVEPOINT *savepoints;
     THD_TRANS all;			// Trans since BEGIN WORK

--- 1.143/sql/sql_delete.cc	2005-03-22 08:25:36 +01:00
+++ 1.144/sql/sql_delete.cc	2005-04-22 20:28:44 +02:00
@@ -70,8 +70,6 @@
 
   select_lex->no_error= thd->lex->ignore;
 
-  table->file->ha_stmt_begin();
-
   /*
     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
@@ -237,8 +235,6 @@
   delete select;
   transactional_table= table->file->has_transactions();
  
-  table->file->ha_stmt_end();
-
   /*
     We write to the binary log even if we deleted no row, because maybe the
     user is using this command to ensure that a table is clean on master *and

--- 1.145/sql/sql_insert.cc	2005-03-22 08:25:36 +01:00
+++ 1.146/sql/sql_insert.cc	2005-04-22 20:28:44 +02:00
@@ -313,8 +313,6 @@
                            (MODE_STRICT_TRANS_TABLES |
                             MODE_STRICT_ALL_TABLES)));
 
-  table->file->ha_stmt_begin();
-
   if (fields.elements && check_that_all_fields_are_given_values(thd, table))
   {
     /* thd->net.report_error is now set, which will abort the next loop */
@@ -405,8 +403,6 @@
     if (table->triggers)
       table->triggers->process_triggers(thd, TRG_EVENT_INSERT, TRG_ACTION_AFTER);
   }
-
-  table->file->ha_stmt_end();
 
   /*
     Now all rows are inserted.  Time to update logs and sends response to

--- 1.151/sql/sql_update.cc	2005-03-22 08:25:38 +01:00
+++ 1.152/sql/sql_update.cc	2005-04-22 20:28:44 +02:00
@@ -398,8 +398,6 @@
                                (MODE_STRICT_TRANS_TABLES |
                                 MODE_STRICT_ALL_TABLES)));
 
-  table->file->ha_stmt_begin();
-
   while (!(error=info.read_record(&info)) && !thd->killed)
   {
     if (!(select && select->skip_record()))
@@ -462,8 +460,6 @@
   delete select;
   thd->proc_info="end";
   VOID(table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY));
-
-  table->file->ha_stmt_end();
 
   /*
     Invalidate the table in the query cache if something changed.