Merge kindahl-laptop.dnsalias.net:/home/bkroot/mysql-5.1-rpl

into  kindahl-laptop.dnsalias.net:/home/bk/b23051-mysql-5.1-rpl


BitKeeper/deleted/.del-binlog_row_blackhole.result:
  Auto merged
sql/ha_ndbcluster.cc:
  Auto merged
sql/handler.cc:
  Auto merged
sql/mysql_priv.h:
  Auto merged
sql/sql_base.cc:
  Auto merged
sql/share/errmsg.txt:
  Auto merged
storage/blackhole/ha_blackhole.h:
  Auto merged
storage/innobase/handler/ha_innodb.cc:
  Auto merged
storage/innobase/handler/ha_innodb.h:
  Auto merged
storage/myisam/ha_myisam.cc:
  Auto merged
mysql-test/t/partition_hash.test:
  Manual merge
sql/handler.h:
  Manual merge
sql/set_var.cc:
  Manual merge
sql/sql_class.h:
  Manual merge
sql/sql_insert.cc:
  Manual merge
sql/sql_parse.cc:
  Manual merge

10 files changed, 208 insertions, 45 deletions

diff --git a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc
index afb2589f6b9..cbbae5b9d0d 100644
--- a/sql/ha_ndbcluster.cc
+++ b/sql/ha_ndbcluster.cc
@@ -4509,7 +4509,6 @@ int ha_ndbcluster::external_lock(THD *thd, int lock_type)
       DBUG_PRINT("warning", ("ops_pending != 0L"));
     m_ops_pending= 0;
   }
-  thd->set_current_stmt_binlog_row_based_if_mixed();
   DBUG_RETURN(error);
 }
 
@@ -4559,7 +4558,6 @@ int ha_ndbcluster::start_stmt(THD *thd, thr_lock_type lock_type)
   m_active_trans= trans;
   // Start of statement
   m_ops_pending= 0;    
-  thd->set_current_stmt_binlog_row_based_if_mixed();
 
   DBUG_RETURN(error);
 }
@@ -6117,6 +6115,7 @@ void ha_ndbcluster::get_auto_increment(ulonglong offset, ulonglong increment,
                 HA_PRIMARY_KEY_REQUIRED_FOR_DELETE | \
                 HA_PARTIAL_COLUMN_READ | \
                 HA_HAS_OWN_BINLOGGING | \
+                HA_BINLOG_ROW_CAPABLE | \
                 HA_HAS_RECORDS
 
 ha_ndbcluster::ha_ndbcluster(handlerton *hton, TABLE_SHARE *table_arg):
@@ -8923,7 +8922,6 @@ pthread_handler_t ndb_util_thread_func(void *arg __attribute__((unused)))
   my_net_init(&thd->net, 0);
   thd->main_security_ctx.master_access= ~0;
   thd->main_security_ctx.priv_user = 0;
-  thd->current_stmt_binlog_row_based= TRUE;     // If in mixed mode
 
   /* Signal successful initialization */
   ndb_util_thread_running= 1;
diff --git a/sql/handler.cc b/sql/handler.cc
index 6bdcdc72bdb..456885f3c6f 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -3639,7 +3639,15 @@ int handler::ha_external_lock(THD *thd, int lock_type)
     taken a table lock), ha_release_auto_increment() was too.
   */
   DBUG_ASSERT(next_insert_id == 0);
-  DBUG_RETURN(external_lock(thd, lock_type));
+
+  /*
+    We cache the table flags if the locking succeeded. Otherwise, we
+    keep them as they were when they were fetched in ha_open().
+  */
+  int error= external_lock(thd, lock_type);
+  if (error == 0)
+    cached_table_flags= table_flags();
+  DBUG_RETURN(error);
 }
 
 
diff --git a/sql/handler.h b/sql/handler.h
index 216620a6882..6bd55a9511e 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -117,6 +117,18 @@
 #define HA_HAS_RECORDS	       (LL(1) << 32) /* records() gives exact count*/
 /* Has it's own method of binlog logging */
 #define HA_HAS_OWN_BINLOGGING  (LL(1) << 33)
+/*
+  Engine is capable of row-format and statement-format logging,
+  respectively
+*/
+#define HA_BINLOG_ROW_CAPABLE  (LL(1) << 34)
+#define HA_BINLOG_STMT_CAPABLE (LL(1) << 35)
+
+/*
+  Set of all binlog flags. Currently only contain the capabilities
+  flags.
+ */
+#define HA_BINLOG_FLAGS (HA_BINLOG_ROW_CAPABLE | HA_BINLOG_STMT_CAPABLE)
 
 /* bits in index_flags(index_number) for what you can do with index */
 #define HA_READ_NEXT            1       /* TODO really use this flag */
@@ -688,7 +700,7 @@ struct handlerton
 };
 
 
-/* Possible flags of a handlerton */
+/* Possible flags of a handlerton (there can be 32 of them) */
 #define HTON_NO_FLAGS                 0
 #define HTON_CLOSE_CURSORS_AT_COMMIT (1 << 0)
 #define HTON_ALTER_NOT_SUPPORTED     (1 << 1) //Engine does not support alter
@@ -793,19 +805,37 @@ typedef struct st_key_create_information
 class TABLEOP_HOOKS
 {
 public:
+  TABLEOP_HOOKS() {}
+  virtual ~TABLEOP_HOOKS() {}
+
   inline void prelock(TABLE **tables, uint count)
   {
     do_prelock(tables, count);
   }
-  virtual ~TABLEOP_HOOKS() {}
-  TABLEOP_HOOKS() {}
 
+  inline int postlock(TABLE **tables, uint count)
+  {
+    return do_postlock(tables, count);
+  }
 private:
   /* Function primitive that is called prior to locking tables */
   virtual void do_prelock(TABLE **tables, uint count)
   {
     /* Default is to do nothing */
   }
+
+  /**
+     Primitive called after tables are locked.
+
+     If an error is returned, the tables will be unlocked and error
+     handling start.
+
+     @return Error code or zero.
+   */
+  virtual int do_postlock(TABLE **tables, uint count)
+  {
+    return 0;                           /* Default is to do nothing */
+  }
 };
 
 typedef struct st_savepoint SAVEPOINT;
@@ -892,10 +922,13 @@ class handler :public Sql_alloc
   friend int ha_delete_table(THD*,handlerton*,const char*,const char*,
                              const char*,bool);
 
+public:
+  typedef ulonglong Table_flags;
+
  protected:
   struct st_table_share *table_share;   /* The table definition */
   struct st_table *table;               /* The current open table */
-  ulonglong cached_table_flags;         /* Set on init() and open() */
+  Table_flags cached_table_flags;       /* Set on init() and open() */
 
   virtual int index_init(uint idx, bool sorted) { active_index=idx; return 0; }
   virtual int index_end() { active_index=MAX_KEY; return 0; }
@@ -908,7 +941,7 @@ class handler :public Sql_alloc
   */
   virtual int rnd_init(bool scan) =0;
   virtual int rnd_end() { return 0; }
-  virtual ulonglong table_flags(void) const =0;
+  virtual Table_flags table_flags(void) const =0;
 
   void ha_statistic_increment(ulong SSV::*offset) const;
   void **ha_data(THD *) const;
@@ -1122,7 +1155,7 @@ public:
   {
     return inited == INDEX ? ha_index_end() : inited == RND ? ha_rnd_end() : 0;
   }
-  longlong ha_table_flags() { return cached_table_flags; }
+  Table_flags ha_table_flags() const { return cached_table_flags; }
 
   /*
     Signal that the table->read_set and table->write_set table maps changed
@@ -1686,6 +1719,8 @@ private:
 	/* Some extern variables used with handlers */
 
 extern const char *ha_row_type[];
+extern const char *tx_isolation_names[];
+extern const char *binlog_format_names[];
 extern TYPELIB tx_isolation_typelib;
 extern TYPELIB myisam_stats_method_typelib;
 extern ulong total_ha, total_ha_2pc;
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index f0a4ef45219..ac8a2fbcadd 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -1274,6 +1274,7 @@ int simple_open_n_lock_tables(THD *thd,TABLE_LIST *tables);
 bool open_and_lock_tables(THD *thd,TABLE_LIST *tables);
 bool open_normal_and_derived_tables(THD *thd, TABLE_LIST *tables, uint flags);
 int lock_tables(THD *thd, TABLE_LIST *tables, uint counter, bool *need_reopen);
+int decide_logging_format(THD *thd, TABLE_LIST *tables);
 TABLE *open_temporary_table(THD *thd, const char *path, const char *db,
 			    const char *table_name, bool link_in_list);
 bool rm_temporary_table(handlerton *base, char *path);
diff --git a/sql/set_var.cc b/sql/set_var.cc
index a6fd3ab1242..70cf8ae1cd1 100644
--- a/sql/set_var.cc
+++ b/sql/set_var.cc
@@ -1007,17 +1007,6 @@ bool sys_var_thd_binlog_format::is_readonly() const
     my_error(ER_STORED_FUNCTION_PREVENTS_SWITCH_BINLOG_FORMAT, MYF(0));
     return 1;    
   }
-#ifdef HAVE_NDB_BINLOG
-  /*
-    Cluster does not support changing the binlog format on the fly yet.
-  */
-  LEX_STRING ndb_name= {(char*)STRING_WITH_LEN("ndbcluster")};
-  if (opt_bin_log && plugin_is_ready(&ndb_name, MYSQL_STORAGE_ENGINE_PLUGIN))
-  {
-    my_error(ER_NDB_CANT_SWITCH_BINLOG_FORMAT, MYF(0));
-    return 1;
-  }
-#endif /* HAVE_NDB_BINLOG */
   return sys_var_thd_enum::is_readonly();
 }
 
diff --git a/sql/share/errmsg.txt b/sql/share/errmsg.txt
index 40744d0b07b..9f5d5f8f720 100644
--- a/sql/share/errmsg.txt
+++ b/sql/share/errmsg.txt
@@ -6062,3 +6062,5 @@ ER_NO_PARTITION_FOR_GIVEN_VALUE_SILENT
 ER_BINLOG_UNSAFE_STATEMENT
         eng "Statement is not safe to log in statement format."
         swe "Detta är inte säkert att logga i statement-format."
+ER_BINLOG_LOGGING_IMPOSSIBLE
+        eng "Binary logging not possible. Message: %s"
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 1986c09b2d2..ed58ed21bbd 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -3948,6 +3948,121 @@ static void mark_real_tables_as_free_for_reuse(TABLE_LIST *table)
 }
 
 
+/**
+   Decide on logging format to use for the statement.
+
+   Compute the capabilities vector for the involved storage engines
+   and mask out the flags for the binary log. Right now, the binlog
+   flags only include the capabilities of the storage engines, so this
+   is safe.
+
+   We now have three alternatives that prevent the statement from
+   being loggable:
+
+   1. If there are no capabilities left (all flags are clear) it is
+      not possible to log the statement at all, so we roll back the
+      statement and report an error.
+
+   2. Statement mode is set, but the capabilities indicate that
+      statement format is not possible.
+
+   3. Row mode is set, but the capabilities indicate that row
+      format is not possible.
+
+   4. Statement is unsafe, but the capabilities indicate that row
+      format is not possible.
+
+   If we are in MIXED mode, we then decide what logging format to use:
+
+   1. If the statement is unsafe, row-based logging is used.
+
+   2. If statement-based logging is not possible, row-based logging is
+      used.
+
+   3. Otherwise, statement-based logging is used.
+
+   @param thd    Client thread
+   @param tables Tables involved in the query
+ */
+
+int decide_logging_format(THD *thd, TABLE_LIST *tables)
+{
+  if (mysql_bin_log.is_open() && (thd->options & OPTION_BIN_LOG))
+  {
+    handler::Table_flags binlog_flags= ~handler::Table_flags();
+    for (TABLE_LIST *table= tables; table; table= table->next_global)
+      if (!table->placeholder() && table->lock_type >= TL_WRITE_ALLOW_WRITE)
+      {
+#define FLAGSTR(S,F) ((S) & (F) ? #F " " : "")
+#ifndef DBUG_OFF
+        ulonglong flags= table->table->file->ha_table_flags();
+        DBUG_PRINT("info", ("table: %s; ha_table_flags: %s%s",
+                            table->table_name,
+                            FLAGSTR(flags, HA_BINLOG_STMT_CAPABLE),
+                            FLAGSTR(flags, HA_BINLOG_ROW_CAPABLE)));
+#endif
+        binlog_flags &= table->table->file->ha_table_flags();
+      }
+    binlog_flags&= HA_BINLOG_FLAGS;
+    DBUG_PRINT("info", ("binlog_flags: %s%s",
+                        FLAGSTR(binlog_flags, HA_BINLOG_STMT_CAPABLE),
+                        FLAGSTR(binlog_flags, HA_BINLOG_ROW_CAPABLE)));
+    DBUG_PRINT("info", ("thd->variables.binlog_format: %ld",
+                        thd->variables.binlog_format));
+
+    int error= 0;
+    if (binlog_flags == 0)
+    {
+      my_error((error= ER_BINLOG_LOGGING_IMPOSSIBLE), MYF(0),
+               "Statement cannot be logged to the binary log in"
+               " row-based nor statement-based format");
+    }
+    else if (thd->variables.binlog_format == BINLOG_FORMAT_STMT &&
+             (binlog_flags & HA_BINLOG_STMT_CAPABLE) == 0)
+    {
+      my_error((error= ER_BINLOG_LOGGING_IMPOSSIBLE), MYF(0),
+                "Statement-based format required for this statement,"
+                " but not allowed by this combination of engines");
+    }
+    else if ((thd->variables.binlog_format == BINLOG_FORMAT_ROW ||
+              thd->lex->is_stmt_unsafe()) &&
+             (binlog_flags & HA_BINLOG_ROW_CAPABLE) == 0)
+    {
+      my_error((error= ER_BINLOG_LOGGING_IMPOSSIBLE), MYF(0),
+                "Row-based format required for this statement,"
+                " but not allowed by this combination of engines");
+    }
+
+    DBUG_PRINT("info", ("error: %d", error));
+
+    if (error)
+    {
+      ha_rollback_stmt(thd);
+      return -1;
+    }
+
+    /*
+      We switch to row-based format if we are in mixed mode and one of
+      the following are true:
+
+      1. If the statement is unsafe
+      2. If statement format cannot be used
+
+      Observe that point to cannot be decided before the tables
+      involved in a statement has been checked, i.e., we cannot put
+      this code in reset_current_stmt_binlog_row_based(), it has to be
+      here.
+    */
+    if (thd->lex->is_stmt_unsafe() ||
+        (binlog_flags & HA_BINLOG_STMT_CAPABLE) == 0)
+    {
+      thd->set_current_stmt_binlog_row_based_if_mixed();
+    }
+  }
+
+  return 0;
+}
+
 /*
   Lock all tables in list
 
@@ -3986,17 +4101,10 @@ int lock_tables(THD *thd, TABLE_LIST *tables, uint count, bool *need_reopen)
     in prelocked mode.
   */
   DBUG_ASSERT(!thd->prelocked_mode || !thd->lex->requires_prelocking());
-
   *need_reopen= FALSE;
 
-  /*
-    CREATE ... SELECT UUID() locks no tables, we have to test here.
-  */
-  if (thd->lex->is_stmt_unsafe())
-    thd->set_current_stmt_binlog_row_based_if_mixed();
-
   if (!tables && !thd->lex->requires_prelocking())
-    DBUG_RETURN(0);
+    DBUG_RETURN(decide_logging_format(thd, tables));
 
   /*
     We need this extra check for thd->prelocked_mode because we want to avoid
@@ -4049,6 +4157,7 @@ int lock_tables(THD *thd, TABLE_LIST *tables, uint count, bool *need_reopen)
       }
       DBUG_RETURN(-1);
     }
+
     if (thd->lex->requires_prelocking() &&
         thd->lex->sql_command != SQLCOM_LOCK_TABLES)
     {
@@ -4115,7 +4224,8 @@ int lock_tables(THD *thd, TABLE_LIST *tables, uint count, bool *need_reopen)
       thd->prelocked_mode= PRELOCKED_UNDER_LOCK_TABLES;
     }
   }
-  DBUG_RETURN(0);
+
+  DBUG_RETURN(decide_logging_format(thd, tables));
 }
 
 
diff --git a/sql/sql_class.h b/sql/sql_class.h
index fb713d8a80b..4217fa4b350 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -2002,20 +2002,21 @@ class select_insert :public select_result_interceptor {
 class select_create: public select_insert {
   ORDER *group;
   TABLE_LIST *create_table;
+  TABLE_LIST *select_tables;
   HA_CREATE_INFO *create_info;
   Alter_info *alter_info;
   Field **field;
 public:
-  select_create(TABLE_LIST *table_arg,
-                HA_CREATE_INFO *create_info_arg,
-                Alter_info *alter_info_arg,
-                List<Item> &select_fields,
-                enum_duplicates duplic, bool ignore)
-    :select_insert(NULL, NULL, &select_fields, 0, 0, duplic, ignore),
+  select_create (TABLE_LIST *table_arg,
+		 HA_CREATE_INFO *create_info_par,
+		 List<create_field> &fields_par,
+		 List<Key> &keys_par,
+		 List<Item> &select_fields,enum_duplicates duplic, bool ignore,
+                 TABLE_LIST *select_tables_arg)
+    :select_insert (NULL, NULL, &select_fields, 0, 0, duplic, ignore),
     create_table(table_arg),
-    create_info(create_info_arg),
-    alter_info(alter_info_arg)
-  {}
+    create_info(create_info_par), select_tables(select_tables_arg)
+    {}
   int prepare(List<Item> &list, SELECT_LEX_UNIT *u);
 
   void binlog_show_create_table(TABLE **tables, uint count);
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 6be2c95a661..b4fa275bbdd 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -3341,8 +3341,15 @@ static TABLE *create_table_from_items(THD *thd, HA_CREATE_INFO *create_info,
   table->reginfo.lock_type=TL_WRITE;
   hooks->prelock(&table, 1);                    // Call prelock hooks
   if (! ((*lock)= mysql_lock_tables(thd, &table, 1,
-                                    MYSQL_LOCK_IGNORE_FLUSH, &not_used)))
+                                    MYSQL_LOCK_IGNORE_FLUSH, &not_used)) ||
+        hooks->postlock(&table, 1))
   {
+    if (*lock)
+    {
+      mysql_unlock_tables(thd, *lock);
+      *lock= 0;
+    }
+
     if (!create_info->table_existed)
       drop_open_table(thd, table, create_table->db, create_table->table_name);
     DBUG_RETURN(0);
@@ -3377,24 +3384,35 @@ select_create::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
    */
   class MY_HOOKS : public TABLEOP_HOOKS {
   public:
-    MY_HOOKS(select_create *x) : ptr(x) { }
+    MY_HOOKS(select_create *x, TABLE_LIST *create_table,
+             TABLE_LIST *select_tables)
+      : ptr(x), all_tables(*create_table)
+      {
+        all_tables.next_global= select_tables;
+      }
 
   private:
-    virtual void do_prelock(TABLE **tables, uint count)
+    virtual int do_postlock(TABLE **tables, uint count)
     {
+      THD *thd= const_cast<THD*>(ptr->get_thd());
+      if (int error= decide_logging_format(thd, &all_tables))
+        return error;
+
       TABLE const *const table = *tables;
-      if (ptr->get_thd()->current_stmt_binlog_row_based  &&
+      if (thd->current_stmt_binlog_row_based  &&
           !table->s->tmp_table &&
           !ptr->get_create_info()->table_existed)
       {
         ptr->binlog_show_create_table(tables, count);
       }
+      return 0;
     }
 
     select_create *ptr;
+    TABLE_LIST all_tables;
   };
 
-  MY_HOOKS hooks(this);
+  MY_HOOKS hooks(this, create_table, select_tables);
   hook_ptr= &hooks;
 
   unit= u;
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 9222f9f49db..f2d2a543f29 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -2227,7 +2227,8 @@ mysql_execute_command(THD *thd)
                                        &alter_info,
                                        select_lex->item_list,
                                        lex->duplicates,
-                                       lex->ignore)))
+                                       lex->ignore,
+                                       select_tables)))
         {
           /*
             CREATE from SELECT give its SELECT_LEX for SELECT,