Merge with base 5.2

BUILD/compile-pentium64

@@ -4,7 +4,10 @@ path=`dirname $0`
 . "$path/SETUP.sh"
 
 extra_flags="$pentium64_cflags $fast_cflags"
-extra_configs="$pentium_configs $static_link"
+# On CentOS/Fedora Core 10 amd64, there is system libz.so but not
+# libz.a, so need to use bundled zlib when building static
+# binary. Hence we use --with-zlib-dir=bundled
+extra_configs="$pentium_configs $static_link --with-zlib-dir=bundled"
 CC="$CC --pipe"
 strip=yes
 

configure.in

@@ -12,6 +12,7 @@ dnl
 dnl When changing the major version number please also check the switch
 dnl statement in mysqlbinlog::check_master_version().  You may also need
 dnl to update version.c in ndb.
+
 AC_INIT([MariaDB Server], [5.2.5-MariaDB], [], [mysql])
 
 AC_CONFIG_SRCDIR([sql/mysqld.cc])

mysql-test/README

@@ -4,13 +4,14 @@ this directory. It will fire up the newly built mysqld and test it.
 
 Note that you do not have to have to do "make install", and you could
 actually have a co-existing MySQL installation. The tests will not
-conflict with it.
+conflict with it. To run the test suite in a source directory, you
+must do make first.
 
 All tests must pass. If one or more of them fail on your system, please
 read the following manual section for instructions on how to report the
 problem:
 
-http://dev.mysql.com/doc/mysql/en/mysql-test-suite.html
+http://kb.askmonty.org/v/reporting-bugs
 
 If you want to use an already running MySQL server for specific tests,
 use the --extern option to mysql-test-run. Please note that in this mode,
@@ -27,7 +28,6 @@ With no test cases named on the command line, mysql-test-run falls back
 to the normal "non-extern" behavior. The reason for this is that some
 tests cannot run with an external server.
 
-
 You can create your own test cases. To create a test case, create a new
 file in the t subdirectory using a text editor. The file should have a .test
 extension. For example:
@@ -67,7 +67,12 @@ extension. For example:
    edit the test result to the correct results so that we can verify
    that the bug is corrected in future releases.
 
-To submit your test case, put your .test file and .result file(s) into
+If you want to submit your test case you can send it 
-a tar.gz archive, add a README that explains the problem, ftp the 
+to maria-developers@lists.launchpad.com or attach it to a bug report on
-archive to ftp://support.mysql.com/pub/mysql/secret/ and send a mail
+https://bugs.launchpad.net/maria/.
-to bugs@lists.mysql.com
+
+If the test case is really big or if it contains 'not public' data,
+then put your .test file and .result file(s) into a tar.gz archive,
+add a README that explains the problem, ftp the archive to
+ftp://ftp.askmonty.org/private and send a mail to
+https://bugs.launchpad.net/maria/ about it.

mysql-test/extra/rpl_tests/rpl_stop_slave.test

@@ -42,6 +42,7 @@ send STOP SLAVE SQL_THREAD;
 connection slave1;
 --echo # To resume slave SQL thread
 SET DEBUG_SYNC= 'now SIGNAL signal.continue';
+SET DEBUG_SYNC= 'now WAIT_FOR signal.continued';
 SET DEBUG_SYNC= 'RESET';
 
 --echo

mysql-test/include/long_test.inc

@@ -0,0 +1,4 @@
+# We use this --source include to mark a test as taking long to run.
+# We can use this to schedule such test early (to not be left with
+# only one or two long tests running, and rests of works idle), or to
+# run a quick test skipping long-running test cases.

mysql-test/lib/mtr_cases.pm

@@ -89,6 +89,20 @@ sub init_pattern {
 }
 
 
+sub testcase_sort_order {
+  my ($a, $b, $sort_criteria)= @_;
+  my $a_sort_criteria= $sort_criteria->{$a->fullname()};
+  my $b_sort_criteria= $sort_criteria->{$b->fullname()};
+  my $res= $a_sort_criteria cmp $b_sort_criteria;
+  return $res if $res;
+  # Run slow tests first, trying to avoid getting stuck at the end
+  # with a slow test in one worker and the other workers idle.
+  return -1 if $a->{'long_test'} && !$b->{'long_test'};
+  return 1 if !$a->{'long_test'} && $b->{'long_test'};
+
+  return $a->fullname() cmp $b->fullname();
+}
+
 ##############################################################################
 #
 #  Collect information about test cases to be run
@@ -177,9 +191,7 @@ sub collect_test_cases ($$$) {
       $sort_criteria{$tinfo->fullname()} = join(" ", @criteria);
     }
 
-    @$cases = sort {
+    @$cases = sort { testcase_sort_order($a, $b, \%sort_criteria) } @$cases;
-      $sort_criteria{$a->fullname()} . $a->fullname() cmp
-	$sort_criteria{$b->fullname()} . $b->fullname() } @$cases;
 
     # For debugging the sort-order
     # foreach my $tinfo (@$cases)
@@ -1065,6 +1077,7 @@ my @tags=
  ["include/have_example_plugin.inc", "example_plugin_test", 1],
  ["include/have_oqgraph_engine.inc", "oqgraph_test", 1],
  ["include/have_ssl.inc", "need_ssl", 1],
+ ["include/long_test.inc", "long_test", 1],
 );
 
 

mysql-test/mysql-test-run.pl

@@ -729,9 +729,11 @@ sub run_test_server ($$$) {
 	    last;
 	  }
 
-	  # Second best choice is the first that does not fulfill
+	  # From secondary choices, we prefer to pick a 'long-running' test if
-	  # any of the above conditions
+          # possible; this helps avoid getting stuck with a few of those at the
-	  if (!defined $second_best){
+          # end of high --parallel runs, with most workers being idle.
+	  if (!defined $second_best ||
+              ($t->{'long_test'} && !($tests->[$second_best]{'long_test'}))){
 	    #mtr_report("Setting second_best to $i");
 	    $second_best= $i;
 	  }

mysql-test/r/subselect.result

@@ -4749,4 +4749,15 @@ sum(a)	sub
 1	3
 deallocate prepare stmt1;
 drop table t1,t2;
+#
+# Bug LP#693935/#58727: Assertion failure with 
+# a single row subquery returning more than one row
+#
+create table t1 (a char(1) charset utf8);
+insert into t1 values ('a'), ('b');
+create table t2 (a binary(1));
+insert into t2 values ('x'), ('y');
+select * from t2 where a=(select a from t1) and a='x';
+ERROR 21000: Subquery returns more than 1 row
+drop table t1,t2;
 End of 5.1 tests

mysql-test/suite/federated/federated_debug.test

@@ -1,4 +1,5 @@
 --source include/have_debug.inc
+--source include/long_test.inc
 --source federated.inc
 
 --echo #

mysql-test/suite/innodb/r/innodb_mysql.result

@@ -2617,6 +2617,13 @@ rows	3
 Extra	Using index
 DROP TABLE t1;
 #
+# ALTER TABLE IGNORE didn't ignore duplicates for unique add index
+#
+create table t1 (a int primary key, b int) engine = innodb;
+insert into t1 values (1,1),(2,1);
+alter ignore table t1 add unique `main` (b);
+drop table t1;
+#
 End of 5.1 tests
 #
 # Test for bug #39932 "create table fails if column for FK is in different

mysql-test/suite/innodb/t/innodb_mysql.test

@@ -840,6 +840,15 @@ CREATE INDEX b ON t1(a,b,c,d);
 
 DROP TABLE t1;
 
+--echo #
+--echo # ALTER TABLE IGNORE didn't ignore duplicates for unique add index
+--echo #
+
+create table t1 (a int primary key, b int) engine = innodb;
+insert into t1 values (1,1),(2,1);
+alter ignore table t1 add unique `main` (b);
+drop table t1;
+
 --echo #
 
 

mysql-test/suite/maria/r/maria.result

@@ -2624,3 +2624,19 @@ KEY (v3)
 INSERT INTO t1 ( f1 , f2 , f3 , f4 ) SELECT f1 , f4 , f1 , f4 FROM t1;
 DELETE FROM t1;
 drop table t1;
+create table t1 (a int not null primary key, b blob) engine=maria transactional=1;
+insert into t1 values(1,repeat('a',8000));
+insert into t1 values(2,repeat('b',8000));
+insert into t1 values(3,repeat('c',8000));
+flush tables;
+delete from t1 where a>1;
+insert into t1 values(1,repeat('d',8000*3));
+ERROR 23000: Duplicate entry '1' for key 'PRIMARY'
+flush tables;
+check table t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+repair table t1 extended;
+Table	Op	Msg_type	Msg_text
+test.t1	repair	status	OK
+drop table t1;

mysql-test/suite/maria/t/maria-recovery-rtree-ft.test

@@ -6,6 +6,7 @@
 # Binary must be compiled with debug for crash to occur
 --source include/have_debug.inc
 --source include/have_maria.inc
+--source include/long_test.inc
 
 set global aria_log_file_size=4294967295;
 let $MARIA_LOG=.;

mysql-test/suite/maria/t/maria.test

@@ -1910,6 +1910,24 @@ INSERT INTO t1 ( f1 , f2 , f3 , f4 ) SELECT f1 , f4 , f1 , f4 FROM t1;
 DELETE FROM t1;
 drop table t1;
 
+#
+# Test of problem where REPAIR finds old deleted rows.
+#
+
+create table t1 (a int not null primary key, b blob) engine=maria transactional=1;
+insert into t1 values(1,repeat('a',8000));
+insert into t1 values(2,repeat('b',8000));
+insert into t1 values(3,repeat('c',8000));
+flush tables;
+delete from t1 where a>1;
+--error 1062
+insert into t1 values(1,repeat('d',8000*3));
+flush tables;
+check table t1;
+# This failed by finding 2 extra rows.
+repair table t1 extended;
+drop table t1;
+
 #
 # End of test
 #

mysql-test/suite/parts/t/partition_alter2_1_myisam.test

@@ -22,6 +22,8 @@
 # any of the variables.
 #
 
+--source include/long_test.inc
+
 #------------------------------------------------------------------------------#
 # General not engine specific settings and requirements
 

mysql-test/suite/parts/t/partition_alter2_2_myisam.test

@@ -22,6 +22,8 @@
 # any of the variables.
 #
 
+--source include/long_test.inc
+
 #------------------------------------------------------------------------------#
 # General not engine specific settings and requirements
 

mysql-test/suite/parts/t/partition_basic_innodb.test

@@ -22,6 +22,8 @@
 # any of the variables.
 #
 
+--source include/long_test.inc
+
 #------------------------------------------------------------------------------#
 # General not engine specific settings and requirements
 

mysql-test/suite/parts/t/partition_decimal_myisam.test

@@ -22,6 +22,8 @@
 # any of the variables.
 #
 
+--source include/long_test.inc
+
 #------------------------------------------------------------------------------#
 # General not engine specific settings and requirements
 

mysql-test/suite/parts/t/partition_float_myisam.test

@@ -22,6 +22,8 @@
 # any of the variables.
 #
 
+--source include/long_test.inc
+
 #------------------------------------------------------------------------------#
 # General not engine specific settings and requirements
 

mysql-test/suite/parts/t/partition_int_myisam.test

@@ -22,6 +22,8 @@
 # any of the variables.
 #
 
+--source include/long_test.inc
+
 #------------------------------------------------------------------------------#
 # General not engine specific settings and requirements
 

mysql-test/suite/rpl/r/rpl_stop_slave.result

@@ -38,6 +38,7 @@ STOP SLAVE SQL_THREAD;
 [ On Slave1 ]
 # To resume slave SQL thread
 SET DEBUG_SYNC= 'now SIGNAL signal.continue';
+SET DEBUG_SYNC= 'now WAIT_FOR signal.continued';
 SET DEBUG_SYNC= 'RESET';
 
 [ On Slave ]
@@ -63,6 +64,7 @@ STOP SLAVE SQL_THREAD;
 [ On Slave1 ]
 # To resume slave SQL thread
 SET DEBUG_SYNC= 'now SIGNAL signal.continue';
+SET DEBUG_SYNC= 'now WAIT_FOR signal.continued';
 SET DEBUG_SYNC= 'RESET';
 
 [ On Slave ]
@@ -89,6 +91,7 @@ STOP SLAVE SQL_THREAD;
 [ On Slave1 ]
 # To resume slave SQL thread
 SET DEBUG_SYNC= 'now SIGNAL signal.continue';
+SET DEBUG_SYNC= 'now WAIT_FOR signal.continued';
 SET DEBUG_SYNC= 'RESET';
 
 [ On Slave ]
@@ -115,6 +118,7 @@ STOP SLAVE SQL_THREAD;
 [ On Slave1 ]
 # To resume slave SQL thread
 SET DEBUG_SYNC= 'now SIGNAL signal.continue';
+SET DEBUG_SYNC= 'now WAIT_FOR signal.continued';
 SET DEBUG_SYNC= 'RESET';
 
 [ On Slave ]

mysql-test/suite/rpl/t/rpl_deadlock_innodb.test

@@ -7,5 +7,6 @@
 ########################################################
 -- source include/not_ndb_default.inc
 -- source include/have_innodb.inc
+-- source include/long_test.inc
 let $engine_type=innodb;
 -- source extra/rpl_tests/rpl_deadlock.test

mysql-test/suite/rpl/t/rpl_row_sp003.test

@@ -10,6 +10,7 @@
 -- source include/have_binlog_format_row.inc
 # Slow test, don't run during staging part
 -- source include/not_staging.inc
+--source include/long_test.inc
 -- source include/master-slave.inc
 
 let $engine_type=INNODB;

mysql-test/t/join_outer.test

@@ -1,3 +1,5 @@
+--source include/long_test.inc
+
 #
 # test of left outer join
 #

mysql-test/t/multi_update2.test

@@ -1,3 +1,5 @@
+--source include/long_test.inc
+
 #
 # Test of update statement that uses many tables.
 #

mysql-test/t/pool_of_threads.test

@@ -4,6 +4,7 @@
 -- source include/have_pool_of_threads.inc
 # Slow test, don't run during staging part
 -- source include/not_staging.inc
+-- source include/long_test.inc
 -- source include/common-tests.inc
 
 

mysql-test/t/query_cache.test

@@ -1,4 +1,5 @@
 -- source include/have_query_cache.inc
+-- source include/long_test.inc
 
 #
 # Tests with query cache

mysql-test/t/subselect.test

@@ -3759,4 +3759,19 @@ deallocate prepare stmt1;
 
 drop table t1,t2;
 
+--echo #
+--echo # Bug LP#693935/#58727: Assertion failure with 
+--echo # a single row subquery returning more than one row
+--echo #
+
+create table t1 (a char(1) charset utf8);
+insert into t1 values ('a'), ('b');
+create table t2 (a binary(1));
+insert into t2 values ('x'), ('y');
+
+-- error ER_SUBQUERY_NO_1_ROW
+select * from t2 where a=(select a from t1) and a='x';
+
+drop table t1,t2;
+
 --echo End of 5.1 tests

sql/item.h

@@ -574,10 +574,17 @@ public:
   Field *make_string_field(TABLE *table);
   virtual bool fix_fields(THD *, Item **);
   /*
-    should be used in case where we are sure that we do not need
+    This method should be used in case where we are sure that we do not need
     complete fix_fields() procedure.
+    Usually this method is used by the optimizer when it has to create a new
+    item out of other already fixed items. For example, if the optimizer has
+    to create a new Item_func for an inferred equality whose left and right
+    parts are already fixed items. In some cases the optimizer cannot use
+    directly fixed items as the arguments of the created functional item, 
+    but rather uses intermediate type conversion items. Then the method is
+    supposed to be applied recursively.  
   */
-  inline void quick_fix_field() { fixed= 1; }
+  virtual inline void quick_fix_field() { fixed= 1; }
   /* Function returns 1 on overflow and -1 on fatal errors */
   int save_in_field_no_warnings(Field *field, bool no_conversions);
   virtual int save_in_field(Field *field, bool no_conversions);

sql/item_func.cc

@@ -202,6 +202,21 @@ Item_func::fix_fields(THD *thd, Item **ref)
   return FALSE;
 }
 
+void
+Item_func::quick_fix_field()
+{
+  Item **arg,**arg_end;
+  if (arg_count)
+  {
+    for (arg=args, arg_end=args+arg_count; arg != arg_end ; arg++)
+    {
+      if (!(*arg)->fixed)
+        (*arg)->quick_fix_field();
+    }
+  }
+  fixed= 1;
+}
+
 
 bool Item_func::walk(Item_processor processor, bool walk_subquery,
                      uchar *argument)

sql/item_func.h

@@ -117,6 +117,7 @@ public:
   // Constructor used for Item_cond_and/or (see Item comment)
   Item_func(THD *thd, Item_func *item);
   bool fix_fields(THD *, Item **ref);
+  void quick_fix_field();
   table_map used_tables() const;
   table_map not_null_tables() const;
   void update_used_tables();

sql/sql_parse.cc

@@ -3233,12 +3233,17 @@ end_with_restore_list:
 
     DBUG_EXECUTE_IF("after_mysql_insert",
                     {
-                      const char act[]=
+                      const char act1[]=
                         "now "
                         "wait_for signal.continue";
+                      const char act2[]=
+                        "now "
+                        "signal signal.continued";
                       DBUG_ASSERT(opt_debug_sync_timeout > 0);
-                      DBUG_ASSERT(!debug_sync_set_action(current_thd,
+                      DBUG_ASSERT(!debug_sync_set_action(thd,
-                                                         STRING_WITH_LEN(act)));
+                                                         STRING_WITH_LEN(act1)));
+                      DBUG_ASSERT(!debug_sync_set_action(thd,
+                                                         STRING_WITH_LEN(act2)));
                     };);
     break;
   }

sql/sql_select.cc

@@ -11214,6 +11214,11 @@ create_internal_tmp_table_from_heap2(THD *thd, TABLE *table,
     DBUG_EXECUTE_IF("raise_error", write_err= HA_ERR_FOUND_DUPP_KEY ;);
     if (write_err)
       goto err;
+    if (thd->killed)
+    {
+      thd->send_kill_message();
+      goto err_killed;
+    }
   }
   /* copy row that filled HEAP table */
   if ((write_err=new_table.file->ha_write_row(table->record[0])))
@@ -11244,6 +11249,7 @@ create_internal_tmp_table_from_heap2(THD *thd, TABLE *table,
  err:
   DBUG_PRINT("error",("Got error: %d",write_err));
   table->file->print_error(write_err, MYF(0));
+err_killed:
   (void) table->file->ha_rnd_end();
   (void) new_table.file->close();
  err1:

sql/sql_table.cc

@@ -7228,6 +7228,16 @@ view_err:
           /* Non-primary unique key. */
           needed_online_flags|=  HA_ONLINE_ADD_UNIQUE_INDEX;
           needed_fast_flags|= HA_ONLINE_ADD_UNIQUE_INDEX_NO_WRITES;
+          if (ignore)
+          {
+            /*
+              If ignore is used, we have to remove all duplicate rows,
+              which require a full table copy.
+            */
+            need_copy_table= ALTER_TABLE_DATA_CHANGED;
+            pk_changed= 2;                      // Don't change need_copy_table
+            break;
+          }
         }
       }
       else

storage/maria/ha_maria.cc

@@ -1059,8 +1059,6 @@ int ha_maria::close(void)
 
 int ha_maria::write_row(uchar * buf)
 {
-  ha_statistic_increment(&SSV::ha_write_count);
-
   /* If we have a timestamp column, update it to the current time */
   if (table->timestamp_field_type & TIMESTAMP_AUTO_SET_ON_INSERT)
     table->timestamp_field->set_time();
@@ -2131,7 +2129,6 @@ bool ha_maria::is_crashed() const
 int ha_maria::update_row(const uchar * old_data, uchar * new_data)
 {
   CHECK_UNTIL_WE_FULLY_IMPLEMENTED_VERSIONING("UPDATE in WRITE CONCURRENT");
-  ha_statistic_increment(&SSV::ha_update_count);
   if (table->timestamp_field_type & TIMESTAMP_AUTO_SET_ON_UPDATE)
     table->timestamp_field->set_time();
   return maria_update(file, old_data, new_data);
@@ -2141,7 +2138,6 @@ int ha_maria::update_row(const uchar * old_data, uchar * new_data)
 int ha_maria::delete_row(const uchar * buf)
 {
   CHECK_UNTIL_WE_FULLY_IMPLEMENTED_VERSIONING("DELETE in WRITE CONCURRENT");
-  ha_statistic_increment(&SSV::ha_delete_count);
   return maria_delete(file, buf);
 }
 
@@ -2151,7 +2147,6 @@ int ha_maria::index_read_map(uchar * buf, const uchar * key,
 			     enum ha_rkey_function find_flag)
 {
   DBUG_ASSERT(inited == INDEX);
-  ha_statistic_increment(&SSV::ha_read_key_count);
   int error= maria_rkey(file, buf, active_index, key, keypart_map, find_flag);
   table->status= error ? STATUS_NOT_FOUND : 0;
   return error;
@@ -2162,7 +2157,6 @@ int ha_maria::index_read_idx_map(uchar * buf, uint index, const uchar * key,
 				 key_part_map keypart_map,
 				 enum ha_rkey_function find_flag)
 {
-  ha_statistic_increment(&SSV::ha_read_key_count);
   int error= maria_rkey(file, buf, index, key, keypart_map, find_flag);
   table->status= error ? STATUS_NOT_FOUND : 0;
   return error;
@@ -2174,7 +2168,6 @@ int ha_maria::index_read_last_map(uchar * buf, const uchar * key,
 {
   DBUG_ENTER("ha_maria::index_read_last_map");
   DBUG_ASSERT(inited == INDEX);
-  ha_statistic_increment(&SSV::ha_read_key_count);
   int error= maria_rkey(file, buf, active_index, key, keypart_map,
                         HA_READ_PREFIX_LAST);
   table->status= error ? STATUS_NOT_FOUND : 0;
@@ -2185,7 +2178,6 @@ int ha_maria::index_read_last_map(uchar * buf, const uchar * key,
 int ha_maria::index_next(uchar * buf)
 {
   DBUG_ASSERT(inited == INDEX);
-  ha_statistic_increment(&SSV::ha_read_next_count);
   int error= maria_rnext(file, buf, active_index);
   table->status= error ? STATUS_NOT_FOUND : 0;
   return error;
@@ -2195,7 +2187,6 @@ int ha_maria::index_next(uchar * buf)
 int ha_maria::index_prev(uchar * buf)
 {
   DBUG_ASSERT(inited == INDEX);
-  ha_statistic_increment(&SSV::ha_read_prev_count);
   int error= maria_rprev(file, buf, active_index);
   table->status= error ? STATUS_NOT_FOUND : 0;
   return error;
@@ -2205,7 +2196,6 @@ int ha_maria::index_prev(uchar * buf)
 int ha_maria::index_first(uchar * buf)
 {
   DBUG_ASSERT(inited == INDEX);
-  ha_statistic_increment(&SSV::ha_read_first_count);
   int error= maria_rfirst(file, buf, active_index);
   table->status= error ? STATUS_NOT_FOUND : 0;
   return error;
@@ -2215,7 +2205,6 @@ int ha_maria::index_first(uchar * buf)
 int ha_maria::index_last(uchar * buf)
 {
   DBUG_ASSERT(inited == INDEX);
-  ha_statistic_increment(&SSV::ha_read_last_count);
   int error= maria_rlast(file, buf, active_index);
   table->status= error ? STATUS_NOT_FOUND : 0;
   return error;
@@ -2228,7 +2217,6 @@ int ha_maria::index_next_same(uchar * buf,
 {
   int error;
   DBUG_ASSERT(inited == INDEX);
-  ha_statistic_increment(&SSV::ha_read_next_count);
   /*
     TODO: Delete this loop in Maria 1.5 as versioning will ensure this never
     happens
@@ -2260,7 +2248,6 @@ int ha_maria::rnd_end()
 
 int ha_maria::rnd_next(uchar *buf)
 {
-  ha_statistic_increment(&SSV::ha_read_rnd_next_count);
   int error= maria_scan(file, buf);
   table->status= error ? STATUS_NOT_FOUND : 0;
   return error;
@@ -2282,7 +2269,6 @@ int ha_maria::restart_rnd_next(uchar *buf)
 
 int ha_maria::rnd_pos(uchar *buf, uchar *pos)
 {
-  ha_statistic_increment(&SSV::ha_read_rnd_count);
   int error= maria_rrnd(file, buf, my_get_ptr(pos, ref_length));
   table->status= error ? STATUS_NOT_FOUND : 0;
   return error;

storage/maria/ma_bitmap.c

@@ -399,7 +399,8 @@ my_bool _ma_bitmap_flush_all(MARIA_SHARE *share)
       become false, wake them up.
     */
     DBUG_PRINT("info", ("bitmap flusher waking up others"));
-    pthread_cond_broadcast(&bitmap->bitmap_cond);
+    if (bitmap->flush_all_requested)
+      pthread_cond_broadcast(&bitmap->bitmap_cond);
   }
   pthread_mutex_unlock(&bitmap->bitmap_lock);
   DBUG_RETURN(res);
@@ -465,7 +466,8 @@ void _ma_bitmap_unlock(MARIA_SHARE *share)
   bitmap->flush_all_requested--;
   bitmap->non_flushable= 0;
   pthread_mutex_unlock(&bitmap->bitmap_lock);
-  pthread_cond_broadcast(&bitmap->bitmap_cond);
+  if (bitmap->flush_all_requested > 0)
+    pthread_cond_broadcast(&bitmap->bitmap_cond);
   DBUG_VOID_RETURN;
 }
 

storage/maria/ma_blockrec.c

@@ -2506,7 +2506,7 @@ static my_bool free_full_page_range(MARIA_HA *info, pgcache_page_no_t page,
   }
   if (delete_count &&
       pagecache_delete_pages(share->pagecache, &info->dfile,
-                             page, delete_count, PAGECACHE_LOCK_WRITE, 0))
+                             page, delete_count, PAGECACHE_LOCK_WRITE, 1))
     res= 1;
 
   if (share->now_transactional)
@@ -2816,7 +2816,6 @@ static my_bool write_block_record(MARIA_HA *info,
   DBUG_PRINT("info", ("Used head length on page: %u  header_length: %u",
                       head_length,
                       (uint) (flag & ROW_FLAG_TRANSID ? TRANSID_SIZE : 0)));
-  DBUG_ASSERT(data <= end_of_data);
   if (head_length < share->base.min_block_length)
   {
     /* Extend row to be of size min_block_length */
@@ -2825,6 +2824,7 @@ static my_bool write_block_record(MARIA_HA *info,
     data+= diff_length;
     head_length= share->base.min_block_length;
   }
+  DBUG_ASSERT(data <= end_of_data);
   /*
     If this is a redo entry (ie, undo_lsn != LSN_ERROR) then we should have
     written exactly head_length bytes (same as original record).
@@ -3492,7 +3492,9 @@ static my_bool allocate_and_write_block_record(MARIA_HA *info,
 
   /* page will be pinned & locked by get_head_or_tail_page */
   if (get_head_or_tail_page(info, blocks->block, info->buff,
-                            row->space_on_head_page, HEAD_PAGE,
+                            max(row->space_on_head_page,
+                                info->s->base.min_block_length),
+                            HEAD_PAGE,
                             PAGECACHE_LOCK_WRITE, &row_pos))
     goto err;
   row->lastpos= ma_recordpos(blocks->block->page, row_pos.rownr);
@@ -4179,6 +4181,13 @@ static my_bool delete_head_or_tail(MARIA_HA *info,
                                 log_data, NULL))
         DBUG_RETURN(1);
     }
+    /*
+      Mark that this page must be written to disk by page cache, even
+      if we could call pagecache_delete() on it.
+      This is needed to ensure that repair finds the empty page on disk
+      and not old data.
+    */
+    pagecache_set_write_on_delete_by_link(page_link.link);
     DBUG_ASSERT(empty_space >= share->bitmap.sizes[0]);
   }
 
@@ -4932,7 +4941,7 @@ int _ma_read_block_record2(MARIA_HA *info, uchar *record,
       goto err;
   }
 #ifdef EXTRA_DEBUG
-  if (share->calc_checksum)
+  if (share->calc_checksum && !info->in_check_table)
   {
     /* Esnure that row checksum is correct */
     DBUG_ASSERT(((share->calc_checksum)(info, record) & 255) ==
@@ -6485,7 +6494,13 @@ err:
    @param  info            Maria handler
    @param  header          Header (without FILEID)
 
-   @note It marks the pages free in the bitmap
+   Mark the pages free in the bitmap.
+
+   We have to check against _ma_redo_not_needed_for_page()
+   to guard against the case where we first clear a block and after
+   that insert new data into the blocks.  If we would unconditionally
+   clear the bitmap here, future changes would be ignored for the page
+   if it's not in the dirty list (ie, it would be flushed).
 
    @return Operation status
      @retval 0      OK
@@ -6494,19 +6509,25 @@ err:
 
 uint _ma_apply_redo_free_blocks(MARIA_HA *info,
                                 LSN lsn __attribute__((unused)),
+                                LSN redo_lsn,
                                 const uchar *header)
 {
   MARIA_SHARE *share= info->s;
   uint ranges;
+  uint16 sid;
   DBUG_ENTER("_ma_apply_redo_free_blocks");
 
   share->state.changed|= (STATE_CHANGED | STATE_NOT_ZEROFILLED |
                           STATE_NOT_MOVABLE);
 
+  sid= fileid_korr(header);
+  header+= FILEID_STORE_SIZE;
   ranges= pagerange_korr(header);
   header+= PAGERANGE_STORE_SIZE;
   DBUG_ASSERT(ranges > 0);
 
+  /** @todo leave bitmap lock to the bitmap code... */
+  pthread_mutex_lock(&share->bitmap.bitmap_lock);
   while (ranges--)
   {
     my_bool res;
@@ -6523,18 +6544,22 @@ uint _ma_apply_redo_free_blocks(MARIA_HA *info,
 
     DBUG_PRINT("info", ("page: %lu  pages: %u", (long) page, page_range));
 
-    /** @todo leave bitmap lock to the bitmap code... */
+    for ( ; page_range-- ; start_page++)
-    pthread_mutex_lock(&share->bitmap.bitmap_lock);
-    res= _ma_bitmap_reset_full_page_bits(info, &share->bitmap, start_page,
-                                         page_range);
-    pthread_mutex_unlock(&share->bitmap.bitmap_lock);
-    if (res)
     {
-      _ma_mark_file_crashed(share);
+      if (_ma_redo_not_needed_for_page(sid, redo_lsn, start_page, FALSE))
-      DBUG_ASSERT(0);
+        continue;
-      DBUG_RETURN(res);
+      res= _ma_bitmap_reset_full_page_bits(info, &share->bitmap, start_page,
+                                           1);
+      if (res)
+      {
+        pthread_mutex_unlock(&share->bitmap.bitmap_lock);
+        _ma_mark_file_crashed(share);
+        DBUG_ASSERT(0);
+        DBUG_RETURN(res);
+      }
     }
   }
+  pthread_mutex_unlock(&share->bitmap.bitmap_lock);
   DBUG_RETURN(0);
 }
 
@@ -6687,21 +6712,23 @@ uint _ma_apply_redo_insert_row_blobs(MARIA_HA *info,
       uint      page_range;
       pgcache_page_no_t page, start_page;
       uchar     *buff;
+      uint	data_on_page= data_size;
 
       start_page= page= page_korr(header);
       header+= PAGE_STORE_SIZE;
       page_range= pagerange_korr(header);
       header+= PAGERANGE_STORE_SIZE;
 
-      for (i= page_range; i-- > 0 ; page++)
+      for (i= page_range; i-- > 0 ; page++, data+= data_on_page)
       {
         MARIA_PINNED_PAGE page_link;
         enum pagecache_page_lock unlock_method;
         enum pagecache_page_pin unpin_method;
-        uint length;
 
         set_if_smaller(first_page2, page);
         set_if_bigger(last_page2, page);
+        if (i == 0 && sub_ranges == 0)
+          data_on_page= data_size - empty_space; /* data on last page */
         if (_ma_redo_not_needed_for_page(sid, redo_lsn, page, FALSE))
           continue;
 
@@ -6764,7 +6791,7 @@ uint _ma_apply_redo_insert_row_blobs(MARIA_HA *info,
                                        PAGECACHE_LOCK_WRITE_UNLOCK,
                                        PAGECACHE_UNPIN, LSN_IMPOSSIBLE,
                                        LSN_IMPOSSIBLE, 0, FALSE);
-              continue;
+              goto fix_bitmap;
             }
             DBUG_ASSERT((found_page_type == (uchar) BLOB_PAGE) ||
                         (found_page_type == (uchar) UNALLOCATED_PAGE));
@@ -6780,33 +6807,32 @@ uint _ma_apply_redo_insert_row_blobs(MARIA_HA *info,
         lsn_store(buff, lsn);
         buff[PAGE_TYPE_OFFSET]= BLOB_PAGE;
 
-        length= data_size;
+        if (data_on_page != data_size)
-        if (i == 0 && sub_ranges == 0)
         {
           /*
             Last page may be only partly filled. We zero the rest, like
             write_full_pages() does.
           */
-          length-= empty_space;
           bzero(buff + share->block_size - PAGE_SUFFIX_SIZE - empty_space,
                 empty_space);
         }
-        memcpy(buff+ PAGE_TYPE_OFFSET + 1, data, length);
+        memcpy(buff+ PAGE_TYPE_OFFSET + 1, data, data_on_page);
-        data+= length;
         if (pagecache_write(share->pagecache,
                             &info->dfile, page, 0,
                             buff, PAGECACHE_PLAIN_PAGE,
                             unlock_method, unpin_method,
                             PAGECACHE_WRITE_DELAY, 0, LSN_IMPOSSIBLE))
           goto err;
-      }
+
+    fix_bitmap:
       /** @todo leave bitmap lock to the bitmap code... */
-      pthread_mutex_lock(&share->bitmap.bitmap_lock);
+        pthread_mutex_lock(&share->bitmap.bitmap_lock);
-      res= _ma_bitmap_set_full_page_bits(info, &share->bitmap, start_page,
+        res= _ma_bitmap_set_full_page_bits(info, &share->bitmap, page,
-                                         page_range);
+                                           1);
-      pthread_mutex_unlock(&share->bitmap.bitmap_lock);
+        pthread_mutex_unlock(&share->bitmap.bitmap_lock);
-      if (res)
+        if (res)
-        goto err;
+          goto err;
+      }
     }
   }
   *first_page= first_page2;

storage/maria/ma_blockrec.h

@@ -235,7 +235,7 @@ uint _ma_apply_redo_insert_row_head_or_tail(MARIA_HA *info, LSN lsn,
 uint _ma_apply_redo_purge_row_head_or_tail(MARIA_HA *info, LSN lsn,
                                            uint page_type,
                                            const uchar *header);
-uint _ma_apply_redo_free_blocks(MARIA_HA *info, LSN lsn,
+uint _ma_apply_redo_free_blocks(MARIA_HA *info, LSN lsn, LSN rec_lsn,
                                 const uchar *header);
 uint _ma_apply_redo_free_head_or_tail(MARIA_HA *info, LSN lsn,
                                       const uchar *header);

storage/maria/ma_check.c

@@ -100,6 +100,9 @@ static my_bool _ma_flush_table_files_before_swap(HA_CHECK *param,
 static TrID max_trid_in_system(void);
 static void _ma_check_print_not_visible_error(HA_CHECK *param, TrID used_trid);
 void retry_if_quick(MARIA_SORT_PARAM *param, int error);
+static void print_bitmap_description(MARIA_SHARE *share,
+                                     pgcache_page_no_t page,
+                                     uchar *buff);
 
 
 /* Initialize check param with default values */
@@ -1842,6 +1845,8 @@ static int check_block_record(HA_CHECK *param, MARIA_HA *info, int extend,
       }
       param->used+= block_size;
       param->link_used+= block_size;
+      if (param->verbose > 2)
+        print_bitmap_description(share, page, bitmap_buff);
       continue;
     }
     /* Skip pages marked as empty in bitmap */
@@ -2034,6 +2039,8 @@ int maria_chk_data_link(HA_CHECK *param, MARIA_HA *info, my_bool extend)
   bzero((char*) param->tmp_key_crc,
         share->base.keys * sizeof(param->tmp_key_crc[0]));
 
+  info->in_check_table= 1;       /* Don't assert on checksum errors */
+
   switch (share->data_file_type) {
   case BLOCK_RECORD:
     error= check_block_record(param, info, extend, record);
@@ -2049,6 +2056,8 @@ int maria_chk_data_link(HA_CHECK *param, MARIA_HA *info, my_bool extend)
     break;
   } /* switch */
 
+  info->in_check_table= 0;
+
   if (error)
     goto err;
 
@@ -2177,12 +2186,17 @@ int maria_chk_data_link(HA_CHECK *param, MARIA_HA *info, my_bool extend)
            llstr(param->del_length, llbuff2));
     printf("Empty space:  %12s    Linkdata:     %10s\n",
            llstr(param->empty, llbuff),llstr(param->link_used, llbuff2));
-    if (param->lost)
+    if (share->data_file_type == BLOCK_RECORD)
-      printf("Lost space:   %12s", llstr(param->lost, llbuff));
-    if (param->max_found_trid)
     {
-      printf("Max trans. id: %11s\n",
+      printf("Full pages:   %12s    Tail count: %12s\n",
-             llstr(param->max_found_trid, llbuff));
+             llstr(param->full_page_count, llbuff),
+             llstr(param->tail_count, llbuff2));
+      printf("Lost space:   %12s\n", llstr(param->lost, llbuff));
+      if (param->max_found_trid)
+      {
+        printf("Max trans. id: %11s\n",
+               llstr(param->max_found_trid, llbuff));
+      }
     }
   }
   my_free(record,MYF(0));
@@ -6799,3 +6813,46 @@ void retry_if_quick(MARIA_SORT_PARAM *sort_param, int error)
     param->testflag|=T_RETRY_WITHOUT_QUICK;
   }
 }
+
+/* Print information about bitmap page */
+
+static void print_bitmap_description(MARIA_SHARE *share,
+                                     pgcache_page_no_t page,
+                                     uchar *bitmap_data)
+{
+  uchar *pos, *end;
+  MARIA_FILE_BITMAP *bitmap= &share->bitmap;
+  uint count=0, dot_printed= 0;
+  char buff[80], last[80];
+
+  printf("Bitmap page %lu\n", (ulong) page);
+  page++;
+  last[0]=0;
+  for (pos= bitmap_data, end= pos+ bitmap->used_size ; pos < end ; pos+= 6)
+  {
+    ulonglong bits= uint6korr(pos);    /* 6 bytes = 6*8/3= 16 patterns */
+    uint i;
+
+    for (i= 0; i < 16 ; i++, bits>>= 3)
+    {
+      if (count > 60)
+      {
+        buff[count]= 0;
+        if (strcmp(buff, last))
+        {
+          memcpy(last, buff, count+1);
+          printf("%8lu: %s\n", (ulong) page - count, buff);
+          dot_printed= 0;
+        }
+        else if (!(dot_printed++))
+          printf("...\n");
+        count= 0;
+      }
+      buff[count++]= '0' + (uint) (bits & 7);
+      page++;
+    }
+  }
+  buff[count]= 0;
+  printf("%8lu: %s\n", (ulong) page - count, buff);
+  fputs("\n", stdout);
+}

storage/maria/ma_pagecache.c

@@ -158,6 +158,7 @@ struct st_pagecache_hash_link
 #define PCBLOCK_IN_FLUSH   16 /* block is in flush operation                 */
 #define PCBLOCK_CHANGED    32 /* block buffer contains a dirty page          */
 #define PCBLOCK_DIRECT_W   64 /* possible direct write to the block          */
+#define PCBLOCK_DEL_WRITE 128 /* should be written on delete                 */
 
 /* page status, returned by find_block */
 #define PAGE_READ               0
@@ -1215,7 +1216,7 @@ static void link_to_file_list(PAGECACHE *pagecache,
   link_changed(block, &pagecache->file_blocks[FILE_HASH(*file)]);
   if (block->status & PCBLOCK_CHANGED)
   {
-    block->status&= ~PCBLOCK_CHANGED;
+    block->status&= ~(PCBLOCK_CHANGED | PCBLOCK_DEL_WRITE);
     block->rec_lsn= LSN_MAX;
     pagecache->blocks_changed--;
     pagecache->global_blocks_changed--;
@@ -3472,6 +3473,31 @@ no_key_cache:					/* Key cache is not used */
 }
 
 
+/*
+  @brief Set/reset flag that page always should be flushed on delete
+
+  @param pagecache      pointer to a page cache data structure
+  @param link           direct link to page (returned by read or write)
+  @param write          write on delete flag value
+
+*/
+
+void pagecache_set_write_on_delete_by_link(PAGECACHE_BLOCK_LINK *block)
+{
+  DBUG_ENTER("pagecache_set_write_on_delete_by_link");
+  DBUG_PRINT("enter", ("fd: %d block 0x%lx  %d -> TRUE",
+                       block->hash_link->file.file,
+                       (ulong) block,
+                       (int) block->status & PCBLOCK_DEL_WRITE));
+  DBUG_ASSERT(block->pins); /* should be pinned */
+  DBUG_ASSERT(block->wlocks); /* should be write locked */
+
+  block->status|= PCBLOCK_DEL_WRITE;
+
+  DBUG_VOID_RETURN;
+}
+
+
 /*
   @brief Delete page from the buffer (common part for link and file/page)
 
@@ -3501,6 +3527,7 @@ static my_bool pagecache_delete_internal(PAGECACHE *pagecache,
   }
   if (block->status & PCBLOCK_CHANGED)
   {
+    flush= (flush || (block->status & PCBLOCK_DEL_WRITE));
     if (flush)
     {
       /* The block contains a dirty page - push it out of the cache */

storage/maria/ma_pagecache.h

@@ -251,6 +251,7 @@ extern void pagecache_unpin(PAGECACHE *pagecache,
 extern void pagecache_unpin_by_link(PAGECACHE *pagecache,
                                     PAGECACHE_BLOCK_LINK *link,
                                     LSN lsn);
+extern void pagecache_set_write_on_delete_by_link(PAGECACHE_BLOCK_LINK *block);
 
 
 /* Results of flush operation (bit field in fact) */

storage/maria/ma_recovery.c

@@ -28,6 +28,7 @@
 #include "trnman.h"
 #include "ma_key_recover.h"
 #include "ma_recovery_util.h"
+#include "hash.h"
 
 struct st_trn_for_recovery /* used only in the REDO phase */
 {
@@ -58,6 +59,7 @@ static ulonglong now; /**< for tracking execution time of phases */
 static int (*save_error_handler_hook)(uint, const char *,myf);
 static uint recovery_warnings; /**< count of warnings */
 static uint recovery_found_crashed_tables;
+HASH tables_to_redo;                          /* For maria_read_log */
 
 #define prototype_redo_exec_hook(R)                                          \
   static int exec_REDO_LOGREC_ ## R(const TRANSLOG_HEADER_BUFFER *rec)
@@ -184,6 +186,21 @@ static void print_preamble()
 }
 
 
+static my_bool table_is_part_of_recovery_set(LEX_STRING *file_name)
+{
+  uint offset =0;
+  if (!tables_to_redo.records)
+    return 1;                                   /* Default, recover table */
+
+  /* Skip base directory */
+  if (file_name->str[0] == '.' &&
+      (file_name->str[1] == '/' || file_name->str[1] == '\\'))
+    offset= 2;
+  /* Only recover if table is in hash */
+  return my_hash_search(&tables_to_redo, (uchar*) file_name->str + offset,
+                        file_name->length - offset) != 0;
+}
+
 /**
    @brief Recovers from the last checkpoint.
 
@@ -1643,8 +1660,8 @@ prototype_redo_exec_hook(REDO_FREE_BLOCKS)
   }
 
   buff= log_record_buffer.str;
-  if (_ma_apply_redo_free_blocks(info, current_group_end_lsn,
+  if (_ma_apply_redo_free_blocks(info, current_group_end_lsn, rec->lsn,
-                                 buff + FILEID_STORE_SIZE))
+                                 buff))
     goto end;
   error= 0;
 end:
@@ -3015,10 +3032,11 @@ static MARIA_HA *get_MARIA_HA_from_REDO_record(const
     page= page_korr(rec->header + FILEID_STORE_SIZE);
     llstr(page, llbuf);
     break;
+  case LOGREC_REDO_FREE_BLOCKS:
     /*
-      For REDO_FREE_BLOCKS, no need to look at dirty pages list: it does not
+      We are checking against the dirty pages in _ma_apply_redo_free_blocks()
-      read data pages, only reads/modifies bitmap page(s) which is cheap.
     */
+    break;
   default:
     break;
   }
@@ -3036,6 +3054,12 @@ static MARIA_HA *get_MARIA_HA_from_REDO_record(const
   share= info->s;
   tprint(tracef, ", '%s'", share->open_file_name.str);
   DBUG_ASSERT(in_redo_phase);
+  if (!table_is_part_of_recovery_set(&share->open_file_name))
+  {
+    tprint(tracef, ", skipped by user\n");
+    return NULL;
+  }
+
   if (cmp_translog_addr(rec->lsn, share->lsn_of_file_id) <= 0)
   {
     /*
@@ -3069,7 +3093,6 @@ static MARIA_HA *get_MARIA_HA_from_REDO_record(const
       REDO_INSERT_ROW_BLOBS will consult list by itself, as it covers several
       pages.
     */
-    tprint(tracef, " page %s", llbuf);
     if (_ma_redo_not_needed_for_page(sid, rec->lsn, page,
                                      index_page_redo_entry))
       return NULL;
@@ -3106,6 +3129,13 @@ static MARIA_HA *get_MARIA_HA_from_UNDO_record(const
   }
   share= info->s;
   tprint(tracef, ", '%s'", share->open_file_name.str);
+
+  if (!table_is_part_of_recovery_set(&share->open_file_name))
+  {
+    tprint(tracef, ", skipped by user\n");
+    return NULL;
+  }
+
   if (cmp_translog_addr(rec->lsn, share->lsn_of_file_id) <= 0)
   {
     tprint(tracef, ", table's LOGREC_FILE_ID has LSN (%lu,0x%lx) more recent"

storage/maria/ma_recovery.h

@@ -30,4 +30,6 @@ int maria_apply_log(LSN lsn, LSN lsn_end, enum maria_apply_log_way apply,
                     FILE *trace_file,
                     my_bool execute_undo_phase, my_bool skip_DDLs,
                     my_bool take_checkpoints, uint *warnings_count);
+/* Table of tables to recover */
+extern HASH tables_to_redo;
 C_MODE_END

storage/maria/ma_recovery_util.c

@@ -129,16 +129,20 @@ my_bool _ma_redo_not_needed_for_page(uint16 shortid, LSN lsn,
       Next 2 bytes: table's short id
       Next 5 bytes: page number
     */
+    char llbuf[22];
     uint64 file_and_page_id=
       (((uint64)((index << 16) | shortid)) << 40) | page;
     struct st_dirty_page *dirty_page= (struct st_dirty_page *)
       hash_search(&all_dirty_pages,
                   (uchar *)&file_and_page_id, sizeof(file_and_page_id));
-    DBUG_PRINT("info", ("in dirty pages list: %d", dirty_page != NULL));
+    DBUG_PRINT("info", ("page %lld in dirty pages list: %d",
+                        (ulonglong) page,
+                        dirty_page != NULL));
     if ((dirty_page == NULL) ||
         cmp_translog_addr(lsn, dirty_page->rec_lsn) < 0)
     {
-      tprint(tracef, ", ignoring because of dirty_pages list\n");
+      tprint(tracef, ", ignoring page %s because of dirty_pages list\n",
+             llstr((ulonglong) page, llbuf));
       return TRUE;
     }
   }

storage/maria/maria_def.h

@@ -570,6 +570,7 @@ struct st_maria_handler
   my_bool was_locked;			/* Was locked in panic */
   my_bool append_insert_at_end;		/* Set if concurrent insert */
   my_bool quick_mode;
+  my_bool in_check_table;                /* We are running check tables */
   /* Marker if key_del_changed */
   /* If info->keyread_buff can't be used for rnext */
   my_bool page_changed;

storage/maria/maria_read_log.c

@@ -213,6 +213,9 @@ static struct my_option my_long_options[] =
   {"silent", 's', "Print less information during apply/undo phase",
    &opt_silent, &opt_silent, 0,
    GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
+  {"tables-to-redo", 'T',
+   "List of tables sepearated with , that we should apply REDO on. Use this if you only want to recover some tables",
+   0, 0, 0, GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
   {"verbose", 'v', "Print more information during apply/undo phase",
    &maria_recovery_verbose, &maria_recovery_verbose, 0,
    GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
@@ -245,7 +248,7 @@ static void print_version(void)
 static void usage(void)
 {
   print_version();
-  puts("Copyright (C) 2007 MySQL AB");
+  puts("Copyright (C) 2007 MySQL AB, 2009-2011 Monty Program Ab");
   puts("This software comes with ABSOLUTELY NO WARRANTY. This is free software,");
   puts("and you are welcome to modify and redistribute it under the GPL license\n");
 
@@ -266,10 +269,18 @@ static void usage(void)
 
 #include <help_end.h>
 
+static uchar* my_hash_get_string(const uchar *record, size_t *length,
+                                my_bool first __attribute__ ((unused)))
+{
+  *length= (size_t) (strcend((const char*) record,',')- (const char*) record);
+  return (uchar*) record;
+}
+
+
 static my_bool
 get_one_option(int optid __attribute__((unused)),
                const struct my_option *opt __attribute__((unused)),
-               char *argument __attribute__((unused)))
+               char *argument)
 {
   switch (optid) {
   case '?':
@@ -278,6 +289,23 @@ get_one_option(int optid __attribute__((unused)),
   case 'V':
     print_version();
     exit(0);
+  case 'T':
+  {
+    char *pos;
+    if (!my_hash_inited(&tables_to_redo))
+    {
+      my_hash_init2(&tables_to_redo, 16, &my_charset_bin,
+                    16, 0, 0, my_hash_get_string, 0, HASH_UNIQUE);
+    }
+    do
+    {
+      pos= strcend(argument, ',');
+      if (pos != argument)                      /* Skip empty strings */
+        my_hash_insert(&tables_to_redo, (uchar*) argument);
+      argument= pos+1;
+    } while (*(pos++));
+    break;
+  }
 #ifndef DBUG_OFF
   case '#':
     DBUG_SET_INITIAL(argument ? argument : default_dbug_option);
@@ -290,6 +318,7 @@ get_one_option(int optid __attribute__((unused)),
 static void get_options(int *argc,char ***argv)
 {
   int ho_error;
+  my_bool need_help= 0;
 
   if ((ho_error=handle_options(argc, argv, my_long_options, get_one_option)))
     exit(ho_error);
@@ -297,8 +326,23 @@ static void get_options(int *argc,char ***argv)
   if (!opt_apply)
     opt_apply_undo= FALSE;
 
-  if (((opt_display_only + opt_apply) != 1) || (*argc > 0))
+  if (*argc > 0)
   {
+    need_help= 1;
+    fprintf(stderr, "Too many arguments given\n");
+  }
+  if ((opt_display_only + opt_apply) != 1)
+  {
+    need_help= 1;
+    fprintf(stderr,
+            "You must use one and only one of the options 'display-only' or "
+            "'apply'\n");
+  }
+
+  if (need_help)
+  {
+    fflush(stderr);
+    need_help =1;
     usage();
     exit(1);
   }