Merge jamppa@bk-internal.mysql.com:/home/bk/mysql-maria

into  a88-113-38-195.elisa-laajakaista.fi:/home/my/bk/mysql-maria.new


configure.in:
  Auto merged
include/my_global.h:
  Auto merged
sql/mysqld.cc:
  Auto merged

configure.in

@@ -770,7 +770,7 @@ AC_HEADER_STDC
 AC_HEADER_SYS_WAIT
 AC_CHECK_HEADERS(fcntl.h float.h floatingpoint.h ieeefp.h limits.h \
  memory.h pwd.h select.h \
- stdlib.h stddef.h \
+ stdlib.h stddef.h sys/stat.h \
  strings.h string.h synch.h sys/mman.h sys/socket.h netinet/in.h arpa/inet.h \
  sys/timeb.h sys/types.h sys/un.h sys/vadvise.h sys/wait.h term.h \
  unistd.h utime.h sys/utime.h termio.h termios.h sched.h crypt.h alloca.h \

include/my_atomic.h

@@ -66,7 +66,16 @@
   v=tmp;
 #endif
 
-#ifdef __GNUC__
+/*
+  transparent_union doesn't work in g++
+  Bug ?
+
+  Darwin's gcc doesn't want to put pointers in a transparent_union
+  when built with -arch ppc64. Complains:
+  warning: 'transparent_union' attribute ignored
+*/
+#if defined(__GNUC__) && !defined(__cplusplus) && \
+      ! (defined(__APPLE__) && defined(_ARCH_PPC64))
 /*
   we want to be able to use my_atomic_xxx functions with
   both signed and unsigned integers. But gcc will issue a warning

include/my_global.h

@@ -431,6 +431,9 @@ C_MODE_END
 #ifdef HAVE_FCNTL_H
 #include <fcntl.h>
 #endif
+#ifdef HAVE_SYS_STAT_H
+#include <sys/stat.h>
+#endif
 #ifdef HAVE_SYS_TIMEB_H
 #include <sys/timeb.h>				/* Avoid warnings on SCO */
 #endif

mysql-test/r/maria-recovery-bitmap.result

+drop database if exists mysqltest;
+create database mysqltest;
+use mysqltest;
+* shut down mysqld, removed logs, restarted it
+use mysqltest;
+create table t1 (a varchar(10000)) engine=maria;
+* TEST of over-allocated bitmap not flushed by checkpoint
+insert into t1 values ("bbbbbbb");
+flush table t1;
+* copied t1 for comparison
+insert into t1 values ("bbbbbbb");
+delete from t1 limit 1;
+set session debug="+d,info,enter,exit,maria_over_alloc_bitmap";
+insert into t1 values ("aaaaaaaaa");
+set global maria_checkpoint_interval=1;
+SET SESSION debug="+d,maria_crash";
+* crashing mysqld intentionally
+set global maria_checkpoint_interval=1;
+ERROR HY000: Lost connection to MySQL server during query
+* recovery happens
+check table t1 extended;
+Table	Op	Msg_type	Msg_text
+mysqltest.t1	check	status	OK
+* testing that checksum after recovery is as expected
+Checksum-check
+ok
+use mysqltest;
+drop database mysqltest_for_comparison;
+drop database mysqltest;

mysql-test/r/maria-recovery.result

+set global maria_log_file_size=4294967296;
 drop database if exists mysqltest;
 create database mysqltest;
 use mysqltest;
@@ -118,6 +119,7 @@ a
 00000000
 00000000
 drop table t1;
+* TEST of two REDOs for same page in one REDO group
 * shut down mysqld, removed logs, restarted it
 use mysqltest;
 CREATE TABLE t1 (
@@ -150,6 +152,7 @@ SELECT LENGTH(b) FROM t1 WHERE i=3;
 LENGTH(b)
 5001
 drop table t1;
+* TEST of INSERT vs state.auto_increment
 * shut down mysqld, removed logs, restarted it
 use mysqltest;
 CREATE TABLE t1 (
@@ -184,6 +187,7 @@ t1	CREATE TABLE `t1` (
   PRIMARY KEY (`i`),
   KEY `c` (`c`)
 ) ENGINE=MARIA AUTO_INCREMENT=5 DEFAULT CHARSET=latin1
+* TEST of UPDATE vs state.auto_increment
 * copied t1 for feeding_recovery
 update t1 set i=15 where c="a";
 flush table t1;

mysql-test/t/maria-recovery-bitmap-master.opt

+--skip-stack-trace --skip-core-file
+

mysql-test/t/maria-recovery-bitmap.test

+# Tests of Maria's recovery of the bitmap pages
+
+--source include/not_embedded.inc
+# Don't test this under valgrind, memory leaks will occur as we crash
+--source include/not_valgrind.inc
+# Binary must be compiled with debug for crash to occur
+--source include/have_debug.inc
+--source include/have_maria.inc
+
+--disable_warnings
+drop database if exists mysqltest;
+--enable_warnings
+create database mysqltest;
+
+# Include scripts can perform SQL. For it to not influence the main test
+# they use a separate connection. This way if they use a DDL it would
+# not autocommit in the main test.
+connect (admin, 127.0.0.1, root,,mysqltest,,);
+--enable_reconnect
+
+connection default;
+use mysqltest;
+--enable_reconnect
+
+-- source include/maria_empty_logs.inc
+let $mms_tables=1;
+create table t1 (a varchar(10000)) engine=maria;
+
+# we want recovery to use the tables as they were at time of crash
+let $mvr_restore_old_snapshot=0;
+# UNDO phase prevents physical comparison, normally,
+# so we'll only use checksums to compare.
+let $mms_compare_physically=0;
+let $mvr_crash_statement= set global maria_checkpoint_interval=1;
+
+--echo * TEST of over-allocated bitmap not flushed by checkpoint
+let $mvr_debug_option="+d,maria_crash";
+insert into t1 values ("bbbbbbb");
+-- source include/maria_make_snapshot_for_comparison.inc
+# make_snapshot_for_comparison closed the table, which lost its id.
+# So we make a null operation just to give a short id to the table so
+# that checkpoint includes table in checkpoint (otherwise nothing to
+# test).
+insert into t1 values ("bbbbbbb");
+delete from t1 limit 1;
+set session debug="+d,info,enter,exit,maria_over_alloc_bitmap";
+send insert into t1 values ("aaaaaaaaa");
+connection admin;
+# Leave time for INSERT to block after modifying bitmap;
+# in the future we should not use sleep but something like
+# debug_sync_point().
+sleep 5;
+# force a checkpoint, which could, if buggy, flush over-allocated
+# bitmap page; as REDO-UNDO was not written, bitmap and data page
+# would be inconsistent. Correct checkpoint will wait until UNDO is
+# written.
+set global maria_checkpoint_interval=1;
+-- source include/maria_verify_recovery.inc
+
+# disabled until pagecache callback framework is coded at which point
+# we can add a get_lsn() callback for bitmaps, fixing the below bug.
+if (0)
+{
+--echo * TEST of bitmap flushed without REDO-UNDO in the log (WAL violation)
+# before crashing we'll flush the bitmap page
+let $mvr_debug_option="+d,maria_flush_bitmap,maria_crash";
+-- source include/maria_make_snapshot_for_comparison.inc
+lock tables t1 write;
+insert into t1 values (REPEAT('a', 6000));
+# bitmap of after-INSERT will be on disk, but data pages will not; if
+# log is not flushed the bitmap is inconsistent with the data.
+-- source include/maria_verify_recovery.inc
+drop table t1;
+}
+
+# clean up everything
+let $mms_purpose=comparison;
+eval drop database mysqltest_for_$mms_purpose;
+drop database mysqltest;

mysql-test/t/maria-recovery.test

@@ -122,6 +122,7 @@ drop table t1;
 # the rewrite was ignored.
 #
 
+--echo * TEST of two REDOs for same page in one REDO group
 -- source include/maria_empty_logs.inc
 let $mms_tables=1;
 CREATE TABLE t1 (
@@ -144,6 +145,7 @@ SELECT LENGTH(b) FROM t1 WHERE i=3;
 drop table t1;
 
 # Test that INSERT's effect on auto-increment is recovered
+--echo * TEST of INSERT vs state.auto_increment
 -- source include/maria_empty_logs.inc
 let $mms_tables=1;
 CREATE TABLE t1 (
@@ -165,6 +167,7 @@ let $mvr_crash_statement= set global maria_checkpoint_interval=1;
 show create table t1;
 
 # Test that UPDATE's effect on auto-increment is recovered
+--echo * TEST of UPDATE vs state.auto_increment
 -- source include/maria_make_snapshot_for_feeding_recovery.inc
 update t1 set i=15 where c="a";
 -- source include/maria_make_snapshot_for_comparison.inc

mysql-test/t/ps_maria.test

@@ -8,12 +8,11 @@
 # NOTE: PLEASE SEE ps_1general.test (bottom) 
 #       BEFORE ADDING NEW TEST CASES HERE !!!
 
+-- source include/have_maria.inc
 set global maria_log_file_size=4294967296;
 
 use test;
 
--- source include/have_maria.inc
-
 let $type= 'MARIA' ;
 -- source include/ps_create.inc
 -- source include/ps_renew.inc

sql/mysqld.cc

@@ -566,7 +566,7 @@ pthread_key(MEM_ROOT**,THR_MALLOC);
 pthread_key(THD*, THR_THD);
 pthread_mutex_t LOCK_mysql_create_db, LOCK_Acl, LOCK_open, LOCK_thread_count,
 		LOCK_mapped_file, LOCK_status, LOCK_global_read_lock,
-		LOCK_error_log, LOCK_uuid_generator,
+		LOCK_error_log,
 		LOCK_delayed_insert, LOCK_delayed_status, LOCK_delayed_create,
 		LOCK_crypt, LOCK_bytes_sent, LOCK_bytes_received,
 	        LOCK_global_system_variables,
@@ -1354,7 +1354,6 @@ static void clean_up_mutexes()
   (void) pthread_mutex_destroy(&LOCK_global_system_variables);
   (void) rwlock_destroy(&LOCK_system_variables_hash);
   (void) pthread_mutex_destroy(&LOCK_global_read_lock);
-  (void) pthread_mutex_destroy(&LOCK_uuid_generator);
   (void) pthread_mutex_destroy(&LOCK_prepared_stmt_count);
   (void) pthread_cond_destroy(&COND_thread_count);
   (void) pthread_cond_destroy(&COND_refresh);
@@ -3105,7 +3104,6 @@ static int init_thread_environment()
   (void) my_rwlock_init(&LOCK_system_variables_hash, NULL);
   (void) pthread_mutex_init(&LOCK_global_read_lock, MY_MUTEX_INIT_FAST);
   (void) pthread_mutex_init(&LOCK_prepared_stmt_count, MY_MUTEX_INIT_FAST);
-  (void) pthread_mutex_init(&LOCK_uuid_generator, MY_MUTEX_INIT_FAST);
 #ifdef HAVE_OPENSSL
   (void) pthread_mutex_init(&LOCK_des_key_file,MY_MUTEX_INIT_FAST);
 #ifndef HAVE_YASSL

storage/maria/Makefile.am

@@ -122,7 +122,8 @@ libmaria_a_SOURCES =	ma_init.c ma_open.c ma_extra.c ma_info.c ma_rkey.c \
 			ma_rt_index.c ma_rt_key.c ma_rt_mbr.c ma_rt_split.c \
 			ma_sp_key.c ma_control_file.c ma_loghandler.c \
 			ma_pagecache.c ma_pagecaches.c \
-			ma_checkpoint.c ma_recovery.c ma_commit.c
+			ma_checkpoint.c ma_recovery.c ma_commit.c \
+			ma_pagecrc.c
 CLEANFILES =		test?.MA? FT?.MA? isam.log ma_test_all ma_rt_test.MA? sp_test.MA?
 
 SUFFIXES = .sh

storage/maria/ha_maria.cc

@@ -2572,7 +2572,7 @@ static int ha_maria_init(void *p)
                     TRANSLOG_PAGE_SIZE, 0) ||
     translog_init(maria_data_root, log_file_size,
                   MYSQL_VERSION_ID, server_id, maria_log_pagecache,
-                  TRANSLOG_DEFAULT_FLAGS) ||
+                  TRANSLOG_DEFAULT_FLAGS, 0) ||
     maria_recover() ||
     ma_checkpoint_init(checkpoint_interval);
   maria_multi_threaded= TRUE;

storage/maria/ma_blockrec.c

@@ -1242,9 +1242,7 @@ static void make_empty_page(MARIA_HA *info, uchar *buff, uint page_type)
   buff[DIR_FREE_OFFSET]=  END_OF_DIR_FREE_LIST;
   int2store(buff + block_size - PAGE_SUFFIX_SIZE - DIR_ENTRY_SIZE,
             PAGE_HEADER_SIZE);
-  if (!(info->s->options & HA_OPTION_PAGE_CHECKSUM))
-    bfill(buff + block_size - KEYPAGE_CHECKSUM_SIZE,
-          KEYPAGE_CHECKSUM_SIZE, (uchar) 255);
+
   DBUG_VOID_RETURN;
 }
 
@@ -1555,10 +1553,6 @@ static my_bool write_full_pages(MARIA_HA *info,
             (data_size - copy_length) + PAGE_SUFFIX_SIZE);
 #endif
 
-    if (!(info->s->options & HA_OPTION_PAGE_CHECKSUM))
-      bfill(buff + block_size - KEYPAGE_CHECKSUM_SIZE,
-            KEYPAGE_CHECKSUM_SIZE, (uchar) 255);
-
     DBUG_ASSERT(share->pagecache->block_size == block_size);
     if (pagecache_write(share->pagecache,
                         &info->dfile, page, 0,
@@ -2033,6 +2027,7 @@ static my_bool write_block_record(MARIA_HA *info,
       break;
     default:                                    /* Wrong data */
       DBUG_ASSERT(0);
+      length=0;
       break;
     }
     if (!tmp_data_used && tmp_data + length > end_of_data)
@@ -2692,32 +2687,21 @@ static my_bool allocate_and_write_block_record(MARIA_HA *info,
   MARIA_BITMAP_BLOCKS *blocks= &row->insert_blocks;
   DBUG_ENTER("allocate_and_write_block_record");
 
+  _ma_bitmap_flushable(info->s, 1);
   if (_ma_bitmap_find_place(info, row, blocks))
-    DBUG_RETURN(1);                         /* Error reading bitmap */
+    goto err;                         /* Error reading bitmap */
 
-#ifdef RECOVERY_EXTRA_DEBUG
-  /* Send this over-allocated bitmap to disk and crash, see if recovers */
-  DBUG_EXECUTE_IF("maria_flush_bitmap",
-                  {
-                    DBUG_PRINT("maria_flush_bitmap", ("now"));
-                    _ma_bitmap_flush(info->s);
-                    _ma_flush_table_files(info, MARIA_FLUSH_DATA |
-                                          MARIA_FLUSH_INDEX,
-                                          FLUSH_KEEP, FLUSH_KEEP);
-                  });
-  DBUG_EXECUTE_IF("maria_crash",
-                  {
-                    DBUG_PRINT("maria_crash", ("now"));
-                    fflush(DBUG_FILE);
-                    abort();
-                  });
-#endif
+  /*
+    Sleep; a checkpoint will happen and should not send this over-allocated
+    bitmap to disk but rather wait.
+  */
+  DBUG_EXECUTE_IF("maria_over_alloc_bitmap", sleep(10););
 
   /* 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,
                             PAGECACHE_LOCK_WRITE, &row_pos))
-    DBUG_RETURN(1);
+    goto err;
   row->lastpos= ma_recordpos(blocks->block->page, row_pos.rownr);
   if (info->s->calc_checksum)
   {
@@ -2732,11 +2716,17 @@ static my_bool allocate_and_write_block_record(MARIA_HA *info,
   if (write_block_record(info, (uchar*) 0, record, row,
                          blocks, blocks->block->org_bitmap_value != 0,
                          &row_pos, undo_lsn, 0))
-    DBUG_RETURN(1);                         /* Error reading bitmap */
+    goto err;                         /* Error reading bitmap */
   DBUG_PRINT("exit", ("Rowid: %lu (%lu:%u)", (ulong) row->lastpos,
                       (ulong) ma_recordpos_to_page(row->lastpos),
                       ma_recordpos_to_dir_entry(row->lastpos)));
+  /* Now let checkpoint happen but don't commit */
+  DBUG_EXECUTE_IF("maria_over_alloc_bitmap", sleep(1000););
   DBUG_RETURN(0);
+err:
+  _ma_bitmap_flushable(info->s, -1);
+  _ma_unpin_all_pages_and_finalize_row(info, LSN_IMPOSSIBLE);
+  DBUG_RETURN(1);
 }
 
 
@@ -2806,6 +2796,7 @@ my_bool _ma_write_abort_block_record(MARIA_HA *info)
   MARIA_SHARE *share= info->s;
   DBUG_ENTER("_ma_write_abort_block_record");
 
+  _ma_bitmap_flushable(share, 1);
   if (delete_head_or_tail(info,
                           ma_recordpos_to_page(info->cur_row.lastpos),
                           ma_recordpos_to_dir_entry(info->cur_row.lastpos), 1,
@@ -2840,6 +2831,7 @@ my_bool _ma_write_abort_block_record(MARIA_HA *info)
                       &lsn, (void*) 0))
       res= 1;
   }
+  _ma_bitmap_flushable(share, -1);
   _ma_unpin_all_pages_and_finalize_row(info, lsn);
   DBUG_RETURN(res);
 }
@@ -2889,12 +2881,13 @@ static my_bool _ma_update_block_record2(MARIA_HA *info,
   calc_record_size(info, record, new_row);
   page= ma_recordpos_to_page(record_pos);
 
+  _ma_bitmap_flushable(share, 1);
   DBUG_ASSERT(share->pagecache->block_size == block_size);
   if (!(buff= pagecache_read(share->pagecache,
                              &info->dfile, (pgcache_page_no_t) page, 0,
                              info->buff, share->page_type,
                              PAGECACHE_LOCK_WRITE, &page_link.link)))
-    DBUG_RETURN(1);
+    goto err;
   page_link.unlock= PAGECACHE_LOCK_WRITE_UNLOCK;
   page_link.changed= 1;
   push_dynamic(&info->pinned_pages, (void*) &page_link);
@@ -2918,7 +2911,7 @@ static my_bool _ma_update_block_record2(MARIA_HA *info,
     if (extend_area_on_page(buff, dir, rownr, share->block_size,
                             new_row->total_length, &org_empty_size,
                             &rec_offset, &length))
-      DBUG_RETURN(1);
+      goto err;
 
     row_pos.buff= buff;
     row_pos.rownr= rownr;
@@ -2980,6 +2973,7 @@ static my_bool _ma_update_block_record2(MARIA_HA *info,
   DBUG_RETURN(res);
 
 err:
+  _ma_bitmap_flushable(share, -1);
   _ma_unpin_all_pages_and_finalize_row(info, LSN_IMPOSSIBLE);
   DBUG_RETURN(1);
 }
@@ -3288,6 +3282,7 @@ my_bool _ma_delete_block_record(MARIA_HA *info, const uchar *record)
   DBUG_PRINT("enter", ("Rowid: %lu (%lu:%u)", (ulong) info->cur_row.lastpos,
                        (ulong) page, record_number));
 
+  _ma_bitmap_flushable(share, 1);
   if (delete_head_or_tail(info, page, record_number, 1, 0) ||
       delete_tails(info, info->cur_row.tail_positions))
     goto err;
@@ -3334,10 +3329,12 @@ my_bool _ma_delete_block_record(MARIA_HA *info, const uchar *record)
 
   }
 
+  _ma_bitmap_flushable(share, -1);
   _ma_unpin_all_pages_and_finalize_row(info, lsn);
   DBUG_RETURN(0);
 
 err:
+  _ma_bitmap_flushable(share, -1);
   _ma_unpin_all_pages_and_finalize_row(info, LSN_IMPOSSIBLE);
   DBUG_RETURN(1);
 }
@@ -5509,10 +5506,14 @@ uint _ma_apply_redo_insert_row_blobs(MARIA_HA *info,
         enum pagecache_page_pin unpin_method;
         uint length;
 
-        if ((page * info->s->block_size) > info->state->data_file_length)
+        if (((page + 1) * info->s->block_size) >
+            info->state->data_file_length)
         {
           /* New page or half written page at end of file */
-          info->state->data_file_length= page * info->s->block_size;
+          DBUG_PRINT("info", ("Enlarging data file from %lu to %lu",
+                              (ulong) info->state->data_file_length,
+                              (ulong) ((page + 1 ) * info->s->block_size)));
+          info->state->data_file_length= (page + 1) * info->s->block_size;
           buff= info->keyread_buff;
           info->keyread_buff_used= 1;
           make_empty_page(info, buff, BLOB_PAGE);
@@ -5540,7 +5541,12 @@ uint _ma_apply_redo_insert_row_blobs(MARIA_HA *info,
                                        LSN_IMPOSSIBLE, 0);
               DBUG_RETURN(my_errno);
             }
-            /* Physical file was too short; Create new page */
+            /*
+              Physical file was too short, create new page. It can be that
+              recovery started with a file with N pages, wrote page N+2 into
+              pagecache (increased data_file_length but not physical file
+              length), now reads page N+1: the read fails.
+            */
             buff= info->keyread_buff;
             info->keyread_buff_used= 1;
             make_empty_page(info, buff, BLOB_PAGE);
@@ -5637,6 +5643,7 @@ my_bool _ma_apply_undo_row_insert(MARIA_HA *info, LSN undo_lsn,
   if (read_row_extent_info(info, buff, rownr))
     DBUG_RETURN(1);
 
+  _ma_bitmap_flushable(share, 1);
   if (delete_head_or_tail(info, page, rownr, 1, 1) ||
       delete_tails(info, info->cur_row.tail_positions))
     goto err;
@@ -5653,6 +5660,7 @@ my_bool _ma_apply_undo_row_insert(MARIA_HA *info, LSN undo_lsn,
 
   res= 0;
 err:
+  _ma_bitmap_flushable(share, -1);
   _ma_unpin_all_pages_and_finalize_row(info, lsn);
   DBUG_RETURN(res);
 }

storage/maria/ma_blockrec.h

@@ -171,6 +171,7 @@ my_bool _ma_compare_block_record(register MARIA_HA *info,
 my_bool _ma_bitmap_init(MARIA_SHARE *share, File file);
 my_bool _ma_bitmap_end(MARIA_SHARE *share);
 my_bool _ma_bitmap_flush(MARIA_SHARE *share);
+my_bool _ma_bitmap_flush_all(MARIA_SHARE *share);
 void _ma_bitmap_reset_cache(MARIA_SHARE *share);
 my_bool _ma_bitmap_find_place(MARIA_HA *info, MARIA_ROW *row,
                               MARIA_BITMAP_BLOCKS *result_blocks);
@@ -198,6 +199,7 @@ my_bool _ma_check_if_right_bitmap_type(MARIA_HA *info,
                                        uint *bitmap_pattern);
 void _ma_bitmap_delete_all(MARIA_SHARE *share);
 int  _ma_bitmap_create_first(MARIA_SHARE *share);
+void _ma_bitmap_flushable(MARIA_SHARE *share, int non_flushable_inc);
 #ifndef DBUG_OFF
 void _ma_print_bitmap(MARIA_FILE_BITMAP *bitmap, uchar *data,
                       ulonglong page);

storage/maria/ma_check.c

@@ -2727,6 +2727,20 @@ int maria_sort_index(HA_CHECK *param, register MARIA_HA *info, char *name)
 } /* maria_sort_index */
 
 
+/**
+  @brief put CRC on the page
+
+  @param buff            reference on the page buffer.
+  @param pos             position of the page in the file.
+  @param length          length of the page
+*/
+
+static void put_crc(char *buff, my_off_t pos, MARIA_SHARE *share)
+{
+  maria_page_crc_set_index(buff, pos / share->block_size, (uchar*) share);
+}
+
+
 	 /* Sort records recursive using one index */
 
 static int sort_one_index(HA_CHECK *param, MARIA_HA *info,
@@ -2804,6 +2818,7 @@ static int sort_one_index(HA_CHECK *param, MARIA_HA *info,
   /* Fill block with zero and write it to the new index file */
   length= _ma_get_page_used(share, buff);
   bzero((uchar*) buff+length,keyinfo->block_length-length);
+  put_crc(buff, new_page_pos, share);
   if (my_pwrite(new_file,(uchar*) buff,(uint) keyinfo->block_length,
 		new_page_pos,MYF(MY_NABP | MY_WAIT_IF_FULL)))
   {
@@ -4862,9 +4877,13 @@ static int sort_insert_key(MARIA_SORT_PARAM *sort_param,
                           DFLT_INIT_HITS, anc_buff))
       DBUG_RETURN(1);
   }
-  else if (my_pwrite(share->kfile.file, anc_buff,
-		     (uint) keyinfo->block_length,filepos, param->myf_rw))
-    DBUG_RETURN(1);
+  else
+  {
+    put_crc(anc_buff, filepos, share);
+    if (my_pwrite(share->kfile.file, anc_buff,
+                  (uint) keyinfo->block_length, filepos, param->myf_rw))
+      DBUG_RETURN(1);
+  }
   DBUG_DUMP("buff", anc_buff, _ma_get_page_used(share, anc_buff));
 
 	/* Write separator-key to block in next level */
@@ -4982,9 +5001,13 @@ int _ma_flush_pending_blocks(MARIA_SORT_PARAM *sort_param)
                             DFLT_INIT_HITS, key_block->buff))
 	DBUG_RETURN(1);
     }
-    else if (my_pwrite(info->s->kfile.file, key_block->buff,
-		       (uint) keyinfo->block_length,filepos, myf_rw))
+    else
+    {
+      put_crc(key_block->buff, filepos, info->s);
+      if (my_pwrite(info->s->kfile.file, key_block->buff,
+                    (uint) keyinfo->block_length,filepos, myf_rw))
       DBUG_RETURN(1);
+    }
     DBUG_DUMP("buff",key_block->buff,length);
     nod_flag=1;
   }
@@ -5571,6 +5594,14 @@ my_bool create_new_data_handle(MARIA_SORT_PARAM *param, File new_file)
     DBUG_RETURN(1);
 
   new_info= sort_info->new_info;
+  pagecache_file_init(new_info->s->bitmap.file, &maria_page_crc_check_bitmap,
+                      (new_info->s->options & HA_OPTION_PAGE_CHECKSUM ?
+                       &maria_page_crc_set_normal :
+                       &maria_page_filler_set_bitmap), new_info->s);
+  pagecache_file_init(new_info->dfile, &maria_page_crc_check_data,
+                      (new_info->s->options & HA_OPTION_PAGE_CHECKSUM ?
+                       &maria_page_crc_set_normal :
+                       &maria_page_filler_set_normal), new_info->s);
   change_data_file_descriptor(new_info, new_file);
   maria_lock_database(new_info, F_EXTRA_LCK);
   if ((sort_info->param->testflag & T_UNPACK) &&
@@ -5829,7 +5860,7 @@ static int write_log_record_for_repair(const HA_CHECK *param, MARIA_HA *info)
 {
   MARIA_SHARE *share= info->s;
   /* in case this is maria_chk or recovery... */
-  if (translog_inited && !maria_in_recovery &&
+  if (translog_status == TRANSLOG_OK && !maria_in_recovery &&
       share->base.born_transactional)
   {
     my_bool save_now_transactional= share->now_transactional;

storage/maria/ma_commit.c

@@ -51,12 +51,6 @@ int ma_commit(TRN *trn)
     So we need to go the first way.
   */
 
-  /**
-     @todo RECOVERY share's state is written to disk only in
-     maria_lock_database(), so COMMIT record is not the last record of the
-     transaction! It is probably an issue. Recovery of the state is a problem
-     not yet solved.
-  */
   /*
     We do not store "thd->transaction.xid_state.xid" for now, it will be
     needed only when we support XA.

storage/maria/ma_create.c

@@ -727,7 +727,8 @@ int maria_create(const char *name, enum data_file_type datafile_type,
   /* max_data_file_length and max_key_file_length are recalculated on open */
   if (tmp_table)
     share.base.max_data_file_length= (my_off_t) ci->data_file_length;
-  else if (ci->transactional && translog_inited && !maria_in_recovery)
+  else if (ci->transactional && translog_status == TRANSLOG_OK &&
+           !maria_in_recovery)
   {
     /*
       we have checked translog_inited above, because maria_chk may call us
@@ -1048,6 +1049,10 @@ int maria_create(const char *name, enum data_file_type datafile_type,
       DROP+CREATE happened (applying REDOs to the wrong table).
     */
     share.kfile.file= file;
+    pagecache_file_init(share.kfile, &maria_page_crc_check_index,
+                        (share.options & HA_OPTION_PAGE_CHECKSUM ?
+                         &maria_page_crc_set_index :
+                         &maria_page_filler_set_normal), &share);
     if (_ma_update_create_rename_lsn_sub(&share, lsn, FALSE))
       goto err;
     my_free(log_data, MYF(0));
@@ -1240,6 +1245,10 @@ int _ma_initialize_data_file(MARIA_SHARE *share, File dfile)
   {
     share->bitmap.block_size= share->base.block_size;
     share->bitmap.file.file = dfile;
+    pagecache_file_init(share->bitmap.file, &maria_page_crc_check_bitmap,
+                        (share->options & HA_OPTION_PAGE_CHECKSUM ?
+                         &maria_page_crc_set_normal :
+                         &maria_page_filler_set_bitmap), share);
     return _ma_bitmap_create_first(share);
   }
   /*

storage/maria/ma_init.c

@@ -59,7 +59,7 @@ void maria_end(void)
     ft_free_stopwords();
     ma_checkpoint_end();
     trnman_destroy();
-    if (translog_inited)
+    if (translog_status == TRANSLOG_OK)
       translog_destroy();
     end_pagecache(maria_log_pagecache, TRUE);
     end_pagecache(maria_pagecache, TRUE);

storage/maria/ma_key_recover.c

@@ -175,7 +175,7 @@ my_bool write_hook_for_clr_end(enum translog_record_type type
 
 
 /**
-  @brief write hook for undo key insert
+  @brief write hook for undo key
 */
 
 my_bool write_hook_for_undo_key(enum translog_record_type type,

storage/maria/ma_loghandler.h

@@ -59,7 +59,6 @@ struct st_maria_handler;
 /* Length of CRC at end of pages */
 #define ROW_EXTENT_PAGE_SIZE	5
 #define ROW_EXTENT_COUNT_SIZE   2
-#define CRC_LENGTH 4
 /* Size of file id in logs */
 #define FILEID_STORE_SIZE 2
 /* Size of page reference in log */
@@ -255,9 +254,18 @@ C_MODE_START
 #define LOGREC_FIXED_RECORD_2LSN_EXAMPLE 5
 #define LOGREC_VARIABLE_RECORD_2LSN_EXAMPLE 6
 
-extern my_bool translog_init(const char *directory, uint32 log_file_max_size,
-			     uint32 server_version, uint32 server_id,
-			     PAGECACHE *pagecache, uint flags);
+extern void translog_example_table_init();
+extern void translog_table_init();
+#define translog_init(D,M,V,I,C,F,R) \
+  translog_init_with_table(D,M,V,I,C,F,R,&translog_table_init)
+extern my_bool translog_init_with_table(const char *directory,
+                                        uint32 log_file_max_size,
+                                        uint32 server_version,
+                                        uint32 server_id,
+                                        PAGECACHE *pagecache,
+                                        uint flags,
+                                        my_bool readonly,
+                                        void (*init_table_func)());
 
 extern my_bool
 translog_write_record(LSN *lsn, enum translog_record_type type, TRN *trn,
@@ -303,7 +311,14 @@ extern void translog_deassign_id_from_share(struct st_maria_share *share);
 extern void
 translog_assign_id_to_share_from_recovery(struct st_maria_share *share,
                                           uint16 id);
-extern my_bool translog_inited;
+enum enum_translog_status
+{
+  TRANSLOG_UNINITED, /* no initialization done or error during initialization */
+  TRANSLOG_OK,       /* transaction log is functioning */
+  TRANSLOG_READONLY, /* read only mode due to write errors */
+  TRANSLOG_SHUTDOWN  /* going to shutdown the loghandler */
+};
+extern enum enum_translog_status translog_status;
 
 /*
   all the rest added because of recovery; should we make

storage/maria/ma_open.c

@@ -151,6 +151,10 @@ static MARIA_HA *maria_clone_internal(MARIA_SHARE *share, int mode,
   info.errkey= -1;
   info.page_changed=1;
   info.keyread_buff= info.buff + share->base.max_key_block_length;
+  pagecache_file_init(info.dfile, &maria_page_crc_check_data,
+                      (share->options & HA_OPTION_PAGE_CHECKSUM ?
+                       &maria_page_crc_set_normal :
+                       &maria_page_filler_set_normal), share);
   bitmap_init(&info.changed_fields, changed_fields_bitmap,
               share->base.fields, 0);
   if ((*share->init)(&info))
@@ -714,6 +718,10 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
     errpos= 5;
 
     share->kfile.file= kfile;
+    pagecache_file_init(share->kfile, &maria_page_crc_check_index,
+                        (share->options & HA_OPTION_PAGE_CHECKSUM ?
+                         &maria_page_crc_set_index :
+                         &maria_page_filler_set_normal), share);
     share->this_process=(ulong) getpid();
     share->last_process= share->state.process;
     share->base.key_parts=key_parts;
@@ -1062,7 +1070,7 @@ uint _ma_state_info_write(MARIA_SHARE *share, uint pWrite)
   {
     safe_mutex_assert_owner(&share->intern_lock);
   }
-  if (share->base.born_transactional && translog_inited &&
+  if (share->base.born_transactional && translog_status == TRANSLOG_OK &&
       !maria_in_recovery)
   {
     /*
@@ -1100,7 +1108,6 @@ uint _ma_state_info_write(MARIA_SHARE *share, uint pWrite)
 
 uint _ma_state_info_write_sub(File file, MARIA_STATE_INFO *state, uint pWrite)
 {
-  /** @todo RECOVERY write it only at checkpoint time */
   uchar  buff[MARIA_STATE_INFO_SIZE + MARIA_STATE_EXTRA_SIZE];
   uchar *ptr=buff;
   uint	i, keys= (uint) state->header.keys;
@@ -1143,7 +1150,6 @@ uint _ma_state_info_write_sub(File file, MARIA_STATE_INFO *state, uint pWrite)
   {
     mi_sizestore(ptr,state->key_root[i]);		ptr+= 8;
   }
-  /** @todo RECOVERY BUG key_del is a problem for recovery */
   mi_sizestore(ptr,state->key_del);	        	ptr+= 8;
   if (pWrite & 2)				/* From maria_chk */
   {
@@ -1535,6 +1541,14 @@ int _ma_open_datafile(MARIA_HA *info, MARIA_SHARE *share,
   info->dfile.file= share->bitmap.file.file=
     my_open(share->data_file_name, share->mode | O_SHARE,
             MYF(MY_WME));
+  pagecache_file_init(share->bitmap.file, &maria_page_crc_check_bitmap,
+                      (share->options & HA_OPTION_PAGE_CHECKSUM ?
+                       &maria_page_crc_set_normal :
+                       &maria_page_filler_set_bitmap), share);
+  pagecache_file_init(info->dfile, &maria_page_crc_check_data,
+                      (share->options & HA_OPTION_PAGE_CHECKSUM ?
+                       &maria_page_crc_set_normal :
+                       &maria_page_filler_set_normal), share);
   return info->dfile.file >= 0 ? 0 : 1;
 }
 
@@ -1549,6 +1563,10 @@ int _ma_open_keyfile(MARIA_SHARE *share)
   share->kfile.file= my_open(share->unique_file_name,
                              share->mode | O_SHARE,
                              MYF(MY_WME));
+  pagecache_file_init(share->kfile, &maria_page_crc_check_index,
+                      (share->options & HA_OPTION_PAGE_CHECKSUM ?
+                       &maria_page_crc_set_index :
+                       &maria_page_filler_set_normal), share);
   pthread_mutex_unlock(&share->intern_lock);
   return (share->kfile.file < 0);
 }

storage/maria/ma_page.c

@@ -29,7 +29,7 @@ uchar *_ma_fetch_keypage(register MARIA_HA *info,
                          MARIA_PINNED_PAGE **page_link_res)
 {
   uchar *tmp;
-  uint page_size;
+  uint page_size __attribute__((unused));
   MARIA_PINNED_PAGE page_link;
   MARIA_SHARE *share= info->s;
   uint block_size= share->block_size;
@@ -83,7 +83,8 @@ uchar *_ma_fetch_keypage(register MARIA_HA *info,
 
 /* Write a key-page on disk */
 
-int _ma_write_keypage(register MARIA_HA *info, register MARIA_KEYDEF *keyinfo,
+int _ma_write_keypage(register MARIA_HA *info,
+                      register MARIA_KEYDEF *keyinfo __attribute__((unused)),
 		      my_off_t page, enum pagecache_page_lock lock,
                       int level, uchar *buff)
 {
@@ -136,9 +137,6 @@ int _ma_write_keypage(register MARIA_HA *info, register MARIA_KEYDEF *keyinfo,
   }
 #endif
   DBUG_ASSERT(share->pagecache->block_size == block_size);
-  if (!(share->options & HA_OPTION_PAGE_CHECKSUM))
-    bfill(buff + block_size - KEYPAGE_CHECKSUM_SIZE,
-          KEYPAGE_CHECKSUM_SIZE, (uchar) 255);
 
   res= pagecache_write(share->pagecache,
                        &share->kfile, page / block_size,
@@ -247,7 +245,7 @@ int _ma_dispose(register MARIA_HA *info, my_off_t pos, my_bool page_not_read)
                            lock_method, pin_method,
                            PAGECACHE_WRITE_DELAY, &page_link.link,
 			   LSN_IMPOSSIBLE,
-                           0, share->keypage_header+8, 0, 0))
+                           0, share->keypage_header + 8))
     result= 1;
 
 #ifdef IDENTICAL_PAGES_AFTER_RECOVERY

storage/maria/ma_pagecache.c

@@ -15,7 +15,7 @@
    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
 
 /*
-  These functions handle page cacheing for Maria tables.
+  These functions handle page caching for Maria tables.
 
   One cache can handle many files.
   It must contain buffers of the same blocksize.
@@ -601,6 +601,10 @@ static uint pagecache_fwrite(PAGECACHE *pagecache,
 {
   DBUG_ENTER("pagecache_fwrite");
   DBUG_ASSERT(type != PAGECACHE_READ_UNKNOWN_PAGE);
+  /**
+    @todo RECOVERY BUG Here, we should call a callback get_lsn(): it will use
+    lsn_korr() for LSN pages, and translog_get_horizon() for bitmap pages.
+  */
   if (type == PAGECACHE_LSN_PAGE)
   {
     LSN lsn;
@@ -608,8 +612,18 @@ static uint pagecache_fwrite(PAGECACHE *pagecache,
     /* TODO: integrate with page format */
     lsn= lsn_korr(buffer + PAGE_LSN_OFFSET);
     DBUG_ASSERT(LSN_VALID(lsn));
-    translog_flush(lsn);
+    if (translog_flush(lsn))
+      DBUG_RETURN(1);
+  }
+  DBUG_PRINT("info", ("write_callback: 0x%lx  data: 0x%lx",
+                      (ulong) filedesc->write_callback,
+                      (ulong) filedesc->callback_data));
+  if ((filedesc->write_callback)(buffer, pageno, filedesc->callback_data))
+  {
+    DBUG_PRINT("error", ("write callback problem"));
+    DBUG_RETURN(1);
   }
+
   DBUG_RETURN(my_pwrite(filedesc->file, buffer, pagecache->block_size,
                         (pageno)<<(pagecache->shift), flags));
 }
@@ -2398,8 +2412,6 @@ static my_bool make_lock_and_pin(PAGECACHE *pagecache,
       pagecache           pointer to a page cache data structure
       block               block to which buffer the data is to be read
       primary             <-> the current thread will read the data
-      validator           validator of read from the disk data
-      validator_data      pointer to the data need by the validator
 
   RETURN VALUE
     None
@@ -2413,9 +2425,7 @@ static my_bool make_lock_and_pin(PAGECACHE *pagecache,
 
 static void read_block(PAGECACHE *pagecache,
                        PAGECACHE_BLOCK_LINK *block,
-                       my_bool primary,
-                       pagecache_disk_read_validator validator,
-                       uchar* validator_data)
+                       my_bool primary)
 {
 
   /* On entry cache_lock is locked */
@@ -2448,9 +2458,18 @@ static void read_block(PAGECACHE *pagecache,
     else
       block->status= PCBLOCK_READ;
 
-    if (validator != NULL &&
-        (*validator)(block->buffer, validator_data))
+    DBUG_PRINT("info", ("read_callback: 0x%lx  data: 0x%lx",
+                        (ulong) block->hash_link->file.read_callback,
+                        (ulong) block->hash_link->file.callback_data));
+    if ((*block->hash_link->file.read_callback)(block->buffer,
+                                                block->hash_link->pageno,
+                                                block->hash_link->
+                                                file.callback_data))
+    {
+      DBUG_PRINT("error", ("read callback problem"));
       block->status|= PCBLOCK_ERROR;
+    }
+
 
     DBUG_PRINT("read_block",
                ("primary request: new page in cache"));
@@ -2881,25 +2900,20 @@ void pagecache_unpin_by_link(PAGECACHE *pagecache,
 
 
 /*
-  Read a block of data from a cached file into a buffer;
+  @brief Read a block of data from a cached file into a buffer;
 
-  SYNOPSIS
-    pagecache_valid_read()
-    pagecache           pointer to a page cache data structure
-    file                handler for the file for the block of data to be read
-    pageno              number of the block of data in the file
-    level               determines the weight of the data
-    buff                buffer to where the data must be placed
-    type                type of the page
-    lock                lock change
-    link                link to the page if we pin it
-    validator           validator of read from the disk data
-    validator_data      pointer to the data need by the validator
+  @param pagecache      pointer to a page cache data structure
+  @param file           handler for the file for the block of data to be read
+  @param pageno         number of the block of data in the file
+  @param level          determines the weight of the data
+  @param buff           buffer to where the data must be placed
+  @param type           type of the page
+  @param lock           lock change
+  @param link           link to the page if we pin it
 
-  RETURN VALUE
-    Returns address from where the data is placed if successful, 0 - otherwise.
+  @return address from where the data is placed if successful, 0 - otherwise.
 
-    Pin will be chosen according to lock parameter (see lock_to_pin)
+  @note Pin will be chosen according to lock parameter (see lock_to_pin)
 */
 static enum pagecache_page_pin lock_to_pin[2][8]=
 {
@@ -2925,16 +2939,14 @@ static enum pagecache_page_pin lock_to_pin[2][8]=
   }
 };
 
-uchar *pagecache_valid_read(PAGECACHE *pagecache,
-                           PAGECACHE_FILE *file,
-                           pgcache_page_no_t pageno,
-                           uint level,
-                           uchar *buff,
-                           enum pagecache_page_type type,
-                           enum pagecache_page_lock lock,
-                           PAGECACHE_BLOCK_LINK **page_link,
-                           pagecache_disk_read_validator validator,
-                           uchar* validator_data)
+uchar *pagecache_read(PAGECACHE *pagecache,
+                      PAGECACHE_FILE *file,
+                      pgcache_page_no_t pageno,
+                      uint level,
+                      uchar *buff,
+                      enum pagecache_page_type type,
+                      enum pagecache_page_lock lock,
+                      PAGECACHE_BLOCK_LINK **page_link)
 {
   int error= 0;
   enum pagecache_page_pin pin= lock_to_pin[test(buff==0)][lock];
@@ -2991,8 +3003,7 @@ uchar *pagecache_valid_read(PAGECACHE *pagecache,
       DBUG_PRINT("info", ("read block 0x%lx", (ulong)block));
       /* The requested page is to be read into the block buffer */
       read_block(pagecache, block,
-                 (my_bool)(page_st == PAGE_TO_BE_READ),
-                 validator, validator_data);
+                 (my_bool)(page_st == PAGE_TO_BE_READ));
       DBUG_PRINT("info", ("read is done"));
     }
 
@@ -3309,9 +3320,7 @@ my_bool pagecache_write_part(PAGECACHE *pagecache,
                              enum pagecache_write_mode write_mode,
                              PAGECACHE_BLOCK_LINK **page_link,
                              LSN first_REDO_LSN_for_page,
-                             uint offset, uint size,
-                             pagecache_disk_read_validator validator,
-                             uchar* validator_data)
+                             uint offset, uint size)
 {
   PAGECACHE_BLOCK_LINK *block= NULL;
   PAGECACHE_BLOCK_LINK *fake_link;
@@ -3412,12 +3421,20 @@ my_bool pagecache_write_part(PAGECACHE *pagecache,
           memcpy(block->buffer + offset, buff, size);
         block->status= PCBLOCK_READ;
         /*
-          The validator can change the page content (removing page
+          The read_callback can change the page content (removing page
           protection) so it have to be called
         */
-        if (validator != NULL &&
-            (*validator)(block->buffer, validator_data))
+        DBUG_PRINT("info", ("read_callback: 0x%lx  data: 0x%lx",
+                            (ulong) block->hash_link->file.read_callback,
+                            (ulong) block->hash_link->file.callback_data));
+        if ((*block->hash_link->file.read_callback)(block->buffer,
+                                                    block->hash_link->pageno,
+                                                    block->hash_link->
+                                                    file.callback_data))
+        {
+          DBUG_PRINT("error", ("read callback problem"));
           block->status|= PCBLOCK_ERROR;
+        }
         KEYCACHE_DBUG_PRINT("key_cache_insert",
                             ("Page injection"));
 #ifdef THREAD
@@ -3429,7 +3446,6 @@ my_bool pagecache_write_part(PAGECACHE *pagecache,
     }
     else
     {
-      DBUG_ASSERT(validator == 0 && validator_data == 0);
       if (! (block->status & PCBLOCK_CHANGED))
           link_to_changed_list(pagecache, block);
 
@@ -4185,18 +4201,8 @@ my_bool pagecache_collect_changed_blocks_with_lsn(PAGECACHE *pagecache,
       */
       DBUG_ASSERT(block->hash_link != NULL);
       DBUG_ASSERT(block->status & PCBLOCK_CHANGED);
-      /**
-         @todo RECOVERY BUG
-         REDO phase uses PAGECACHE_PLAIN_PAGE, so the lines below would
-         confuse the indirect Checkpoint taken at the end of the REDO phase.
-         So we below collect even dirty pages of temporary tables as a result
-         :( Soon we should have the MARIA_SHARE accessible from the
-         pagecache's block and then we can test born_transactional.
-      */
-#ifdef TRANS_TABLES_ALWAYS_USE_LSN_PAGE
       if (block->type != PAGECACHE_LSN_PAGE)
         continue; /* no need to store it */
-#endif
       stored_list_size++;
     }
   }
@@ -4221,12 +4227,14 @@ my_bool pagecache_collect_changed_blocks_with_lsn(PAGECACHE *pagecache,
          block;
          block= block->next_changed)
     {
-#ifdef TRANS_TABLES_ALWAYS_USE_LSN_PAGE
       if (block->type != PAGECACHE_LSN_PAGE)
         continue; /* no need to store it in the checkpoint record */
-#endif
       compile_time_assert(sizeof(block->hash_link->file.file) <= 4);
       compile_time_assert(sizeof(block->hash_link->pageno) <= 4);
+      /**
+         @todo RECOVERY when we have a pointer to MARIA_SHARE, store share->id
+         instead of this file.
+      */
       int4store(ptr, block->hash_link->file.file);
       ptr+= 4;
       int4store(ptr, block->hash_link->pageno);

storage/maria/ma_pagecache.h

@@ -74,15 +74,20 @@ enum pagecache_write_mode
   PAGECACHE_WRITE_DONE
 };
 
+/* page number for maria */
+typedef uint32 pgcache_page_no_t;
+
 /* file descriptor for Maria */
 typedef struct st_pagecache_file
 {
   File file;
+  my_bool (*read_callback)(uchar *page, pgcache_page_no_t offset,
+                           uchar *data);
+  my_bool (*write_callback)(uchar *page, pgcache_page_no_t offset,
+                            uchar *data);
+  uchar *callback_data;
 } PAGECACHE_FILE;
 
-/* page number for maria */
-typedef uint32 pgcache_page_no_t;
-
 /* declare structures that is used by  st_pagecache */
 
 struct st_pagecache_block_link;
@@ -94,8 +99,6 @@ typedef struct st_pagecache_hash_link PAGECACHE_HASH_LINK;
 
 #include <wqueue.h>
 
-typedef my_bool (*pagecache_disk_read_validator)(uchar *page, uchar *data);
-
 #define PAGECACHE_CHANGED_BLOCKS_HASH 128  /* must be power of 2 */
 #define PAGECACHE_PRIORITY_LOW 0
 #define PAGECACHE_PRIORITY_DEFAULT 3
@@ -192,26 +195,21 @@ extern ulong resize_pagecache(PAGECACHE *pagecache,
 extern void change_pagecache_param(PAGECACHE *pagecache, uint division_limit,
                                    uint age_threshold);
 
-#define pagecache_read(P,F,N,L,B,T,K,I) \
-  pagecache_valid_read(P,F,N,L,B,T,K,I,0,0)
-
-extern uchar *pagecache_valid_read(PAGECACHE *pagecache,
-                                  PAGECACHE_FILE *file,
-                                  pgcache_page_no_t pageno,
-                                  uint level,
-                                  uchar *buff,
-                                  enum pagecache_page_type type,
-                                  enum pagecache_page_lock lock,
-                                  PAGECACHE_BLOCK_LINK **link,
-                                  pagecache_disk_read_validator validator,
-                                  uchar* validator_data);
+extern uchar *pagecache_read(PAGECACHE *pagecache,
+                             PAGECACHE_FILE *file,
+                             pgcache_page_no_t pageno,
+                             uint level,
+                             uchar *buff,
+                             enum pagecache_page_type type,
+                             enum pagecache_page_lock lock,
+                             PAGECACHE_BLOCK_LINK **link);
 
 #define  pagecache_write(P,F,N,L,B,T,O,I,M,K,R) \
-   pagecache_write_part(P,F,N,L,B,T,O,I,M,K,R,0,(P)->block_size,0,0)
+   pagecache_write_part(P,F,N,L,B,T,O,I,M,K,R,0,(P)->block_size)
 
-#define  pagecache_inject(P,F,N,L,B,T,O,I,K,R,V,D) \
+#define  pagecache_inject(P,F,N,L,B,T,O,I,K,R) \
    pagecache_write_part(P,F,N,L,B,T,O,I,PAGECACHE_WRITE_DONE, \
-                        K,R,0,(P)->block_size,V,D)
+                        K,R,0,(P)->block_size)
 
 extern my_bool pagecache_write_part(PAGECACHE *pagecache,
                                     PAGECACHE_FILE *file,
@@ -225,9 +223,7 @@ extern my_bool pagecache_write_part(PAGECACHE *pagecache,
                                     PAGECACHE_BLOCK_LINK **link,
                                     LSN first_REDO_LSN_for_page,
                                     uint offset,
-                                    uint size,
-                                    pagecache_disk_read_validator validator,
-                                    uchar* validator_data);
+                                    uint size);
 extern void pagecache_unlock(PAGECACHE *pagecache,
                              PAGECACHE_FILE *file,
                              pgcache_page_no_t pageno,
@@ -261,6 +257,11 @@ extern void pagecache_unpin_by_link(PAGECACHE *pagecache,
 /* PCFLUSH_ERROR and PCFLUSH_PINNED. */
 #define PCFLUSH_PINNED_AND_ERROR (PCFLUSH_ERROR|PCFLUSH_PINNED)
 
+#define pagecache_file_init(F,RC,WC,D) \
+  do{ \
+    (F).read_callback= (RC); (F).write_callback= (WC); \
+    (F).callback_data= (uchar*)(D); \
+  } while(0)
 
 #define flush_pagecache_blocks(A,B,C)                   \
   flush_pagecache_blocks_with_filter(A,B,C,NULL,NULL)

storage/maria/ma_pagecrc.c

+/* TODO: copyright & Co */
+
+#include "maria_def.h"
+
+
+/**
+  @brief calculate crc of the page avoiding special values
+
+  @param start           The value to start CRC (we use page number here)
+  @param data            data pointer
+  @param length          length of the data
+
+  @return crc of the page without special values
+*/
+
+static uint32 maria_page_crc(ulong start, uchar *data, uint length)
+{
+  uint32 crc= crc32(start, data, length);
+
+  /* we need this assert to get following comparison working */
+  compile_time_assert(MARIA_NO_CRC_BITMAP_PAGE ==
+                      MARIA_NO_CRC_NORMAL_PAGE - 1 &&
+                      MARIA_NO_CRC_NORMAL_PAGE == 0xffffffff);
+  if (crc >= MARIA_NO_CRC_BITMAP_PAGE)
+    crc= MARIA_NO_CRC_BITMAP_PAGE - 1;
+
+  return(crc);
+}
+
+/**
+  @brief Maria pages read callback (checks the page CRC)
+
+  @param page            The page data to check
+  @param page_no         The page number (<offset>/<page length>)
+  @param data_ptr        pointer to MARIA_SHARE
+  @param no_crc_val      Value which means CRC absence
+                         (MARIA_NO_CRC_NORMAL_PAGE or MARIA_NO_CRC_BITMAP_PAGE)
+  @param data_length     length of data to calculate CRC
+
+  @retval 0 OK
+  @retval 1 Error
+*/
+
+static inline my_bool maria_page_crc_check(uchar *page,
+                                           pgcache_page_no_t page_no,
+                                           MARIA_SHARE *share,
+                                           uint32 no_crc_val,
+                                           int data_length)
+{
+  uint32 crc= uint4korr(page + share->block_size - CRC_SIZE), new_crc;
+  my_bool res;
+  DBUG_ENTER("maria_page_crc_check");
+
+  DBUG_ASSERT((uint)data_length <= share->block_size - CRC_SIZE);
+
+  /* we need this assert to get following comparison working */
+  compile_time_assert(MARIA_NO_CRC_BITMAP_PAGE ==
+                      MARIA_NO_CRC_NORMAL_PAGE - 1 &&
+                      MARIA_NO_CRC_NORMAL_PAGE == 0xffffffff);
+  /*
+    If crc is no_crc_val then
+    the page has no crc, so there is nothing to check.
+  */
+  if (crc >= MARIA_NO_CRC_BITMAP_PAGE)
+  {
+    DBUG_PRINT("info", ("No crc: (0x%lx)  crc: (0x%lx) page: %lu  ",
+                        (ulong) no_crc_val, (ulong) crc, (ulong) page_no));
+#ifndef DBUG_OFF
+    if (crc != no_crc_val)
+      DBUG_PRINT("CRCerror", ("Wrong no CRC value"));
+#endif
+    DBUG_RETURN(test(crc != no_crc_val));
+  }
+  new_crc= maria_page_crc(page_no, page, data_length);
+  DBUG_ASSERT(new_crc != no_crc_val);
+  res= test(new_crc != crc);
+  if (res)
+  {
+    DBUG_PRINT("CRCerror", ("Page: %lu  crc: 0x%lx  calculated crc: 0x%lx",
+                            (ulong) page_no, (ulong) crc, (ulong) new_crc));
+    maria_mark_crashed_share(share);
+  }
+  DBUG_RETURN(res);
+}
+
+
+/**
+  @brief Maria pages write callback (sets the page CRC for data and index
+  files)
+
+  @param page            The page data to set
+  @param page_no         The page number (<offset>/<page length>)
+  @param data_ptr        Write callback data pointer (pointer to MARIA_SHARE)
+
+  @retval 0 OK
+*/
+
+my_bool maria_page_crc_set_normal(uchar *page,
+                                  pgcache_page_no_t page_no,
+                                  uchar* data_ptr)
+{
+  MARIA_SHARE *share= (MARIA_SHARE *)data_ptr;
+  int data_length= share->block_size - CRC_SIZE;
+  uint32 crc= maria_page_crc(page_no, page, data_length);
+  DBUG_ENTER("maria_page_crc_set");
+
+  DBUG_PRINT("info", ("Page %u  crc: 0x%lx",
+                      (uint)page_no, (ulong)crc));
+
+  /* crc is on the stack so it is aligned, pagecache buffer is aligned, too */
+  int4store_aligned(page + data_length, crc);
+  DBUG_RETURN(0);
+}
+
+
+/**
+  @brief Maria pages write callback (sets the page CRC for keys)
+
+  @param page            The page data to set
+  @param page_no         The page number (<offset>/<page length>)
+  @param data_ptr        Write callback data pointer (pointer to MARIA_SHARE)
+
+  @retval 0 OK
+*/
+
+my_bool maria_page_crc_set_index(uchar *page,
+                                 pgcache_page_no_t page_no,
+                                 uchar* data_ptr)
+{
+  MARIA_SHARE *share= (MARIA_SHARE *)data_ptr;
+  int data_length= _ma_get_page_used(share, page);
+  uint32 crc= maria_page_crc(page_no, page, data_length);
+  DBUG_ENTER("maria_page_crc_set");
+
+  DBUG_PRINT("info", ("Page %u  crc: 0x%lx",
+                      (uint)page_no, (ulong)crc));
+  DBUG_ASSERT((uint)data_length <= share->block_size - CRC_SIZE);
+  /* crc is on the stack so it is aligned, pagecache buffer is aligned, too */
+  int4store_aligned(page + share->block_size - CRC_SIZE, crc);
+  DBUG_RETURN(0);
+}
+
+
+/* interface functions */
+
+
+/**
+  @brief Maria pages read callback (checks the page CRC) for index/data pages
+
+  @param page            The page data to check
+  @param page_no         The page number (<offset>/<page length>)
+  @param data_ptr        Read callback data pointer (pointer to MARIA_SHARE)
+
+  @retval 0 OK
+  @retval 1 Error
+*/
+
+my_bool maria_page_crc_check_data(uchar *page,
+                                  pgcache_page_no_t page_no,
+                                  uchar* data_ptr)
+{
+  MARIA_SHARE *share= (MARIA_SHARE *)data_ptr;
+  return (maria_page_crc_check(page, page_no, share,
+                               MARIA_NO_CRC_NORMAL_PAGE,
+                               share->block_size - CRC_SIZE));
+}
+
+
+/**
+  @brief Maria pages read callback (checks the page CRC) for bitmap pages
+
+  @param page            The page data to check
+  @param page_no         The page number (<offset>/<page length>)
+  @param data_ptr        Read callback data pointer (pointer to MARIA_SHARE)
+
+  @retval 0 OK
+  @retval 1 Error
+*/
+
+my_bool maria_page_crc_check_bitmap(uchar *page,
+                                    pgcache_page_no_t page_no,
+                                    uchar* data_ptr)
+{
+  MARIA_SHARE *share= (MARIA_SHARE *)data_ptr;
+  return (maria_page_crc_check(page, page_no, share,
+                               MARIA_NO_CRC_BITMAP_PAGE,
+                               share->block_size - CRC_SIZE));
+}
+
+
+/**
+  @brief Maria pages read callback (checks the page CRC) for index pages
+
+  @param page            The page data to check
+  @param page_no         The page number (<offset>/<page length>)
+  @param data_ptr        Read callback data pointer (pointer to MARIA_SHARE)
+
+  @retval 0 OK
+  @retval 1 Error
+*/
+
+my_bool maria_page_crc_check_index(uchar *page,
+                                   pgcache_page_no_t page_no,
+                                   uchar* data_ptr)
+{
+  MARIA_SHARE *share= (MARIA_SHARE *)data_ptr;
+  return (maria_page_crc_check(page, page_no, share,
+                               MARIA_NO_CRC_NORMAL_PAGE,
+                               _ma_get_page_used(share, page)));
+}
+
+
+/**
+  @brief Maria pages write callback (sets the page filler for index/data)
+
+  @param page            The page data to set
+  @param page_no         The page number (<offset>/<page length>)
+  @param data_ptr        Write callback data pointer (pointer to MARIA_SHARE)
+
+  @retval 0 OK
+*/
+
+my_bool maria_page_filler_set_normal(uchar *page,
+                                     __attribute__((unused))
+                                     pgcache_page_no_t page_no,
+                                     uchar* data_ptr)
+{
+  DBUG_ENTER("maria_page_filler_set_normal");
+  int4store_aligned(page + ((MARIA_SHARE *)data_ptr)->block_size - CRC_SIZE,
+                    MARIA_NO_CRC_NORMAL_PAGE);
+  DBUG_RETURN(0);
+}
+
+/**
+  @brief Maria pages write callback (sets the page filler for bitmap)
+
+  @param page            The page data to set
+  @param page_no         The page number (<offset>/<page length>)
+  @param data_ptr        Write callback data pointer (pointer to MARIA_SHARE)
+
+  @retval 0 OK
+*/
+
+my_bool maria_page_filler_set_bitmap(uchar *page,
+                                     __attribute__((unused))
+                                     pgcache_page_no_t page_no,
+                                     uchar* data_ptr)
+{
+  DBUG_ENTER("maria_page_filler_set_bitmap");
+  int4store_aligned(page + ((MARIA_SHARE *)data_ptr)->block_size - CRC_SIZE,
+                    MARIA_NO_CRC_BITMAP_PAGE);
+  DBUG_RETURN(0);
+}

storage/maria/ma_panic.c

@@ -99,12 +99,19 @@ int maria_panic(enum ha_panic_function flag)
       {					/* Open closed files */
 	char name_buff[FN_REFLEN];
 	if (info->s->kfile.file < 0)
+        {
+
 	  if ((info->s->kfile.file= my_open(fn_format(name_buff,
                                                       info->filename, "",
                                                       N_NAME_IEXT,4),
                                             info->mode,
                                             MYF(MY_WME))) < 0)
 	    error = my_errno;
+          pagecache_file_init(info->s->kfile, &maria_page_crc_check_index,
+                              (info->s->options & HA_OPTION_PAGE_CHECKSUM ?
+                               &maria_page_crc_set_index :
+                               &maria_page_filler_set_normal), info->s);
+        }
 	if (info->dfile.file < 0)
 	{
 	  if ((info->dfile.file= my_open(fn_format(name_buff, info->filename,
@@ -112,6 +119,10 @@ int maria_panic(enum ha_panic_function flag)
                                          info->mode,
                                          MYF(MY_WME))) < 0)
 	    error = my_errno;
+          pagecache_file_init(info->dfile, &maria_page_crc_check_data,
+                              (share->options & HA_OPTION_PAGE_CHECKSUM ?
+                               &maria_page_crc_set_normal:
+                               &maria_page_filler_set_normal), share);
 	  info->rec_cache.file= info->dfile.file;
 	}
       }

storage/maria/ma_test1.c

@@ -83,7 +83,7 @@ int main(int argc,char *argv[])
                       TRANSLOG_PAGE_SIZE, MY_WME) == 0) ||
       translog_init(maria_data_root, TRANSLOG_FILE_SIZE,
                     0, 0, maria_log_pagecache,
-                    TRANSLOG_DEFAULT_FLAGS) ||
+                    TRANSLOG_DEFAULT_FLAGS, 0) ||
       (transactional && (trnman_init(0) || ma_checkpoint_init(0))))
   {
     fprintf(stderr, "Error in initialization\n");

storage/maria/ma_test2.c

@@ -96,7 +96,7 @@ int main(int argc, char *argv[])
 		      TRANSLOG_PAGE_SIZE, MY_WME) == 0) ||
       translog_init(maria_data_root, TRANSLOG_FILE_SIZE,
 		    0, 0, maria_log_pagecache,
-		    TRANSLOG_DEFAULT_FLAGS) ||
+		    TRANSLOG_DEFAULT_FLAGS, 0) ||
       (transactional && (trnman_init(0) || ma_checkpoint_init(0))))
   {
     fprintf(stderr, "Error in initialization");

storage/maria/maria_chk.c

@@ -104,8 +104,8 @@ int main(int argc, char **argv)
   maria_init();
 
   /*
-    If we are doing a repair and we have requested logging (on by default),
-    enable transaction log handling.
+    If we are doing a repair, user may want to store this repair into the log
+    so that the log has a complete history and can be used to replay.
   */
   if (opt_transaction_logging && (check_param.testflag & T_REP_ANY) &&
       (ma_control_file_create_or_open() ||
@@ -114,7 +114,7 @@ int main(int argc, char **argv)
                       TRANSLOG_PAGE_SIZE, MY_WME) == 0 ||
        translog_init(maria_data_root, TRANSLOG_FILE_SIZE,
                      0, 0, maria_log_pagecache,
-                     TRANSLOG_DEFAULT_FLAGS)))
+                     TRANSLOG_DEFAULT_FLAGS, 0)))
   {
     _ma_check_print_error(&check_param,
                           "Can't initialize transaction logging. Run "
@@ -1670,6 +1670,10 @@ static int maria_sort_records(HA_CHECK *param,
   VOID(my_close(info->dfile.file, MYF(MY_WME)));
   param->out_flag|=O_NEW_DATA;			/* Data in new file */
   info->dfile.file= new_file;				/* Use new datafile */
+  pagecache_file_init(info->dfile, &maria_page_crc_check_data,
+                      (share->options & HA_OPTION_PAGE_CHECKSUM ?
+                       &maria_page_crc_set_normal :
+                       &maria_page_filler_set_normal), share);
   info->state->del=0;
   info->state->empty=0;
   share->state.dellink= HA_OFFSET_ERROR;

storage/maria/maria_def.h

@@ -44,7 +44,9 @@
 struct st_transaction;
 
 /* undef map from my_nosys; We need test-if-disk full */
-#undef my_write	
+#undef my_write
+
+#define CRC_SIZE 4
 
 typedef struct st_maria_status_info
 {
@@ -217,16 +219,20 @@ typedef struct st_maria_file_bitmap
   ulonglong page;                      /* Page number for current bitmap */
   uint used_size;                      /* Size of bitmap head that is not 0 */
   my_bool changed;                     /* 1 if page needs to be flushed */
+  my_bool flush_all_requested;         /**< If _ma_bitmap_flush_all waiting */
+  uint non_flushable;                  /**< 0 if bitmap and log are in sync */
   PAGECACHE_FILE file;		       /* datafile where bitmap is stored */
 
 #ifdef THREAD
   pthread_mutex_t bitmap_lock;
+  pthread_cond_t bitmap_cond;          /**< When bitmap becomes flushable */
 #endif
   /* Constants, allocated when initiating bitmaps */
   uint sizes[8];                      /* Size per bit combination */
   uint total_size;		      /* Total usable size of bitmap page */
   uint block_size;                    /* Block size of file */
   ulong pages_covered;                /* Pages covered by bitmap + 1 */
+  DYNAMIC_ARRAY pinned_pages;         /**< not-yet-flushable bitmap pages */
 } MARIA_FILE_BITMAP;
 
 #define MARIA_CHECKPOINT_LOOKS_AT_ME 1
@@ -511,7 +517,6 @@ struct st_maria_handler
 
 #define USE_WHOLE_KEY	65535         /* Use whole key in _search() */
 #define F_EXTRA_LCK	-1
-#define TRANSID_SIZE		6
 
 /* bits in opt_flag */
 #define MEMMAP_USED	32
@@ -571,9 +576,19 @@ struct st_maria_handler
 #define _ma_store_keypage_flag(share,x,flag) x[(share)->keypage_header - KEYPAGE_USED_SIZE - KEYPAGE_FLAG_SIZE]= (flag)
 
 
+/*
+  TODO: write int4store_aligned as *((uint32 *) (T))= (uint32) (A) for
+  architectures where it is possible
+*/
+#define int4store_aligned(A,B) int4store((A),(B))
+
 #define maria_mark_crashed(x) do{(x)->s->state.changed|= STATE_CRASHED; \
     DBUG_PRINT("error", ("Marked table crashed"));                      \
   }while(0)
+#define maria_mark_crashed_share(x)                                     \
+  do{(x)->state.changed|= STATE_CRASHED;                                \
+    DBUG_PRINT("error", ("Marked table crashed"));                      \
+  }while(0)
 #define maria_mark_crashed_on_repair(x) do{(x)->s->state.changed|=      \
       STATE_CRASHED|STATE_CRASHED_ON_REPAIR;                            \
     (x)->update|= HA_STATE_CHANGED;                                     \
@@ -1037,4 +1052,27 @@ void _ma_tmp_disable_logging_for_table(MARIA_HA *info,
   { if (((S)->now_transactional= (S)->base.born_transactional))  \
       (S)->page_type= PAGECACHE_LSN_PAGE; }
 
+#define MARIA_NO_CRC_NORMAL_PAGE 0xffffffff
+#define MARIA_NO_CRC_BITMAP_PAGE 0xfffffffe
+extern my_bool maria_page_crc_set_index(uchar *page,
+                                        pgcache_page_no_t page_no,
+                                        uchar* data_ptr);
+extern my_bool maria_page_crc_set_normal(uchar *page,
+                                         pgcache_page_no_t page_no,
+                                         uchar* data_ptr);
+extern my_bool maria_page_crc_check_bitmap(uchar *page,
+                                           pgcache_page_no_t page_no,
+                                           uchar* data_ptr);
+extern my_bool maria_page_crc_check_data(uchar *page,
+                                           pgcache_page_no_t page_no,
+                                           uchar* data_ptr);
+extern my_bool maria_page_crc_check_index(uchar *page,
+                                           pgcache_page_no_t page_no,
+                                           uchar* data_ptr);
+extern my_bool maria_page_filler_set_bitmap(uchar *page,
+                                            pgcache_page_no_t page_no,
+                                            uchar* data_ptr);
+extern my_bool maria_page_filler_set_normal(uchar *page,
+                                            pgcache_page_no_t page_no,
+                                            uchar* data_ptr);
 extern PAGECACHE *maria_log_pagecache;

storage/maria/maria_read_log.c

@@ -82,7 +82,8 @@ int main(int argc, char **argv)
                      TRANSLOG_PAGECACHE_SIZE, 0, 0,
                      TRANSLOG_PAGE_SIZE, MY_WME) == 0 ||
       translog_init(maria_data_root, TRANSLOG_FILE_SIZE,
-                    0, 0, maria_log_pagecache, TRANSLOG_DEFAULT_FLAGS))
+                    0, 0, maria_log_pagecache, TRANSLOG_DEFAULT_FLAGS,
+                    opt_display_only))
   {
     fprintf(stderr, "Can't init loghandler (%d)\n", errno);
     goto err;

storage/maria/trnman.c

@@ -598,6 +598,7 @@ my_bool trnman_collect_transactions(LEX_STRING *str_act, LEX_STRING *str_com,
   pthread_mutex_lock(&LOCK_trn_list);
   str_act->length= 2 + /* number of active transactions */
     LSN_STORE_SIZE + /* minimum of their rec_lsn */
+    TRANSID_SIZE + /* current TrID generator value */
     (2 + /* short id */
      6 + /* long id */
      LSN_STORE_SIZE + /* undo_lsn */
@@ -618,6 +619,8 @@ my_bool trnman_collect_transactions(LEX_STRING *str_act, LEX_STRING *str_com,
     goto err;
   /* First, the active transactions */
   ptr= str_act->str + 2 + LSN_STORE_SIZE;
+  transid_store(ptr, global_trid_generator);
+  ptr+= TRANSID_SIZE;
   for (trn= active_list_min.next; trn != &active_list_max; trn= trn->next)
   {
     /*

storage/maria/trnman_public.h

@@ -55,6 +55,8 @@ my_bool trnman_has_locked_tables(TRN *trn);
 void trnman_reset_locked_tables(TRN *trn);
 TRN *trnman_recreate_trn_from_recovery(uint16 shortid, TrID longid);
 TRN *trnman_get_any_trn();
-
+#define TRANSID_SIZE		6
+#define transid_store(dst, id) int6store(dst,id)
+#define transid_korr(P) uint6korr(P)
 C_MODE_END
 #endif

storage/maria/unittest/Makefile.am

@@ -46,7 +46,9 @@ noinst_PROGRAMS =	ma_control_file-t trnman-t lockman2-t \
 			ma_test_loghandler_noflush-t \
 			ma_test_loghandler_first_lsn-t \
                         ma_test_loghandler_max_lsn-t \
-                        ma_test_loghandler_purge-t
+                        ma_test_loghandler_purge-t \
+			ma_test_loghandler_readonly-t\
+                        ma_test_loghandler_nologs-t
 
 ma_test_loghandler_t_SOURCES = ma_test_loghandler-t.c ma_maria_log_cleanup.c ma_loghandler_examples.c
 ma_test_loghandler_multigroup_t_SOURCES = ma_test_loghandler_multigroup-t.c ma_maria_log_cleanup.c ma_loghandler_examples.c
@@ -58,6 +60,9 @@ ma_test_loghandler_noflush_t_SOURCES = ma_test_loghandler_noflush-t.c ma_maria_l
 ma_test_loghandler_first_lsn_t_SOURCES = ma_test_loghandler_first_lsn-t.c ma_maria_log_cleanup.c ma_loghandler_examples.c
 ma_test_loghandler_max_lsn_t_SOURCES = ma_test_loghandler_max_lsn-t.c ma_maria_log_cleanup.c ma_loghandler_examples.c
 ma_test_loghandler_purge_t_SOURCES = ma_test_loghandler_purge-t.c ma_maria_log_cleanup.c ma_loghandler_examples.c
+ma_test_loghandler_readonly_t_SOURCES = ma_test_loghandler_multigroup-t.c ma_maria_log_cleanup.c ma_loghandler_examples.c
+ma_test_loghandler_readonly_t_CPPFLAGS = -DREADONLY_TEST
+ma_test_loghandler_nologs_t_SOURCES = ma_test_loghandler_nologs-t.c ma_maria_log_cleanup.c ma_loghandler_examples.c
 
 ma_pagecache_single_src =	ma_pagecache_single.c test_file.c test_file.h
 ma_pagecache_consist_src =	ma_pagecache_consist.c test_file.c test_file.h

storage/maria/unittest/ma_loghandler_examples.c

@@ -27,7 +27,7 @@ static LOG_DESC INIT_LOGREC_VARIABLE_RECORD_2LSN_EXAMPLE=
 "variable2example", LOGREC_NOT_LAST_IN_GROUP, NULL, NULL};
 
 
-void example_loghandler_init()
+void translog_example_table_init()
 {
   int i;
   log_record_type_descriptor[LOGREC_FIXED_RECORD_0LSN_EXAMPLE]=

storage/maria/unittest/ma_maria_log_cleanup.c

@@ -37,9 +37,13 @@ my_bool maria_log_remove()
       if (fn_format(file_name, file,
                     maria_data_root, "", MYF(MY_WME)) == NullS ||
           my_delete(file_name, MYF(MY_WME)) != 0)
+      {
+        my_dirend(dirp);
         return 1;
+      }
     }
   }
+  my_dirend(dirp);
   return 0;
 }
 

storage/maria/unittest/ma_pagecache_consist.c

@@ -57,6 +57,18 @@ static uint flush_divider= 1000;
 #endif /*TEST_READERS*/
 #endif /*TEST_HIGH_CONCURENCY*/
 
+/**
+  @brief Dummy pagecache callback.
+*/
+
+static my_bool
+dummy_callback(__attribute__((unused)) uchar *page,
+               __attribute__((unused)) pgcache_page_no_t page_no,
+               __attribute__((unused)) uchar* data_ptr)
+{
+  return 0;
+}
+
 
 /*
   Get pseudo-random length of the field in (0;limit)
@@ -321,6 +333,7 @@ int main(int argc __attribute__((unused)),
 	    errno);
     exit(1);
   }
+  pagecache_file_init(file1, &dummy_callback, &dummy_callback, NULL);
   DBUG_PRINT("info", ("file1: %d", file1.file));
   if (chmod(file1_name, S_IRWXU | S_IRWXG | S_IRWXO) != 0)
   {

storage/maria/unittest/ma_pagecache_single.c

@@ -60,6 +60,19 @@ static struct file_desc  simple_delete_flush_test_file[]=
   { 0, 0}
 };
 
+/**
+  @brief Dummy pagecache callback.
+*/
+
+static my_bool
+dummy_callback(__attribute__((unused)) uchar *page,
+               __attribute__((unused)) pgcache_page_no_t page_no,
+               __attribute__((unused)) uchar* data_ptr)
+{
+  return 0;
+}
+
+
 
 /*
   Recreate and reopen a file for test
@@ -496,7 +509,6 @@ int main(int argc __attribute__((unused)),
 #endif
   DBUG_ENTER("main");
   DBUG_PRINT("info", ("Main thread: %s\n", my_thread_name()));
-
   if ((tmp_file= my_open(file2_name, O_CREAT | O_TRUNC | O_RDWR,
                          MYF(MY_WME))) < 0)
     exit(1);
@@ -508,6 +520,7 @@ int main(int argc __attribute__((unused)),
 	    errno);
     exit(1);
   }
+  pagecache_file_init(file1, &dummy_callback, &dummy_callback, NULL);
   my_close(tmp_file, MYF(0));
   my_delete(file2_name, MYF(0));
 

storage/maria/unittest/ma_test_loghandler-t.c

@@ -176,13 +176,12 @@ int main(int argc __attribute__((unused)), char *argv[])
     fprintf(stderr, "Got error: init_pagecache() (errno: %d)\n", errno);
     exit(1);
   }
-  if (translog_init(".", LOG_FILE_SIZE, 50112, 0, &pagecache, LOG_FLAGS))
+  if (translog_init_with_table(".", LOG_FILE_SIZE, 50112, 0, &pagecache,
+                               LOG_FLAGS, 0, &translog_example_table_init))
   {
     fprintf(stderr, "Can't init loghandler (%d)\n", errno);
-    translog_destroy();
     exit(1);
   }
-  example_loghandler_init();
   /* Suppressing of automatic record writing */
   trn->first_undo_lsn|= TRANSACTION_LOGGED_LONG_ID;
 

storage/maria/unittest/ma_test_loghandler_first_lsn-t.c

@@ -5,7 +5,7 @@
 #include "../trnman.h"
 
 extern my_bool maria_log_remove();
-extern void example_loghandler_init();
+extern void translog_example_table_init();
 
 #ifndef DBUG_OFF
 static const char *default_dbug_option;
@@ -66,13 +66,12 @@ int main(int argc __attribute__((unused)), char *argv[])
     fprintf(stderr, "Got error: init_pagecache() (errno: %d)\n", errno);
     exit(1);
   }
-  if (translog_init(".", LOG_FILE_SIZE, 50112, 0, &pagecache, LOG_FLAGS))
+  if (translog_init_with_table(".", LOG_FILE_SIZE, 50112, 0, &pagecache,
+                               LOG_FLAGS, 0, &translog_example_table_init))
   {
     fprintf(stderr, "Can't init loghandler (%d)\n", errno);
-    translog_destroy();
     exit(1);
   }
-  example_loghandler_init();
   /* Suppressing of automatic record writing */
   dummy_transaction_object.first_undo_lsn|= TRANSACTION_LOGGED_LONG_ID;
 

storage/maria/unittest/ma_test_loghandler_max_lsn-t.c

@@ -5,7 +5,7 @@
 #include "../trnman.h"
 
 extern my_bool maria_log_remove();
-extern void example_loghandler_init();
+extern void translog_example_table_init();
 
 #ifndef DBUG_OFF
 static const char *default_dbug_option;
@@ -60,13 +60,12 @@ int main(int argc __attribute__((unused)), char *argv[])
     fprintf(stderr, "Got error: init_pagecache() (errno: %d)\n", errno);
     exit(1);
   }
-  if (translog_init(".", LOG_FILE_SIZE, 50112, 0, &pagecache, LOG_FLAGS))
+  if (translog_init_with_table(".", LOG_FILE_SIZE, 50112, 0, &pagecache,
+                               LOG_FLAGS, 0, &translog_example_table_init))
   {
     fprintf(stderr, "Can't init loghandler (%d)\n", errno);
-    translog_destroy();
     exit(1);
   }
-  example_loghandler_init();
   /* Suppressing of automatic record writing */
   dummy_transaction_object.first_undo_lsn|= TRANSACTION_LOGGED_LONG_ID;
 

storage/maria/unittest/ma_test_loghandler_multigroup-t.c

@@ -5,22 +5,34 @@
 #include "../trnman.h"
 
 extern my_bool maria_log_remove();
-extern void example_loghandler_init();
+extern void translog_example_table_init();
 
 #ifndef DBUG_OFF
 static const char *default_dbug_option;
 #endif
 static TRN *trn= &dummy_transaction_object;
 
-#define PCACHE_SIZE (1024*1024*10)
 
-#define LONG_BUFFER_SIZE ((1024L*1024L*1024L) + (1024L*1024L*512))
+#ifndef READONLY_TEST
 
+#define PCACHE_SIZE (1024*1024*10)
+#define LONG_BUFFER_SIZE ((1024L*1024L*1024L) + (1024L*1024L*512))
 #define MIN_REC_LENGTH (1024L*1024L + 1024L*512L + 1)
+#define LOG_FILE_SIZE (1024L*1024L*1024L + 1024L*1024L*512)
+#define ITERATIONS 2
+#define READONLY 0
 
+#else
+
+#define PCACHE_SIZE (1024*1024*10)
+#define LONG_BUFFER_SIZE (1024L*1024L)
+#define MIN_REC_LENGTH (1024L)
 #define LOG_FILE_SIZE (1024L*1024L*1024L + 1024L*1024L*512)
 #define ITERATIONS 2
-/*#define ITERATIONS 63 */
+#define READONLY 1
+
+#endif /*READONLY_TEST*/
+
 
 /*
 #define LOG_FILE_SIZE 1024L*1024L*3L
@@ -173,13 +185,12 @@ int main(int argc __attribute__((unused)), char *argv[])
     fprintf(stderr, "Got error: init_pagecache() (errno: %d)\n", errno);
     exit(1);
   }
-  if (translog_init(".", LOG_FILE_SIZE, 50112, 0, &pagecache, 0))
+  if (translog_init_with_table(".", LOG_FILE_SIZE, 50112, 0, &pagecache,
+                               0, 0, &translog_example_table_init))
   {
     fprintf(stderr, "Can't init loghandler (%d)\n", errno);
-    translog_destroy();
     exit(1);
   }
-  example_loghandler_init();
   /* Suppressing of automatic record writing */
   trn->first_undo_lsn|= TRANSACTION_LOGGED_LONG_ID;
 
@@ -338,13 +349,12 @@ int main(int argc __attribute__((unused)), char *argv[])
     fprintf(stderr, "pass2: Got error: init_pagecache() (errno: %d)\n", errno);
     exit(1);
   }
-  if (translog_init(".", LOG_FILE_SIZE, 50112, 0, &pagecache, 0))
+  if (translog_init_with_table(".", LOG_FILE_SIZE, 50112, 0, &pagecache,
+                               0, READONLY, &translog_example_table_init))
   {
     fprintf(stderr, "pass2: Can't init loghandler (%d)\n", errno);
-    translog_destroy();
     exit(1);
   }
-  example_loghandler_init();
 
   srandom(122334817L);
 

storage/maria/unittest/ma_test_loghandler_multithread-t.c

@@ -5,7 +5,7 @@
 #include "../trnman.h"
 
 extern my_bool maria_log_remove();
-extern void example_loghandler_init();
+extern void translog_example_table_init();
 
 #ifndef DBUG_OFF
 static const char *default_dbug_option;
@@ -283,13 +283,12 @@ int main(int argc __attribute__((unused)),
     fprintf(stderr, "Got error: init_pagecache() (errno: %d)\n", errno);
     exit(1);
   }
-  if (translog_init(".", LOG_FILE_SIZE, 50112, 0, &pagecache, LOG_FLAGS))
+  if (translog_init_with_table(".", LOG_FILE_SIZE, 50112, 0, &pagecache,
+                               LOG_FLAGS, 0, &translog_example_table_init))
   {
     fprintf(stderr, "Can't init loghandler (%d)\n", errno);
-    translog_destroy();
     exit(1);
   }
-  example_loghandler_init();
   /* Suppressing of automatic record writing */
   dummy_transaction_object.first_undo_lsn|= TRANSACTION_LOGGED_LONG_ID;
 

storage/maria/unittest/ma_test_loghandler_noflush-t.c

@@ -5,7 +5,7 @@
 #include "../trnman.h"
 
 extern my_bool maria_log_remove();
-extern void example_loghandler_init();
+extern void translog_example_table_init();
 
 #ifndef DBUG_OFF
 static const char *default_dbug_option;
@@ -68,13 +68,12 @@ int main(int argc __attribute__((unused)), char *argv[])
     fprintf(stderr, "Got error: init_pagecache() (errno: %d)\n", errno);
     exit(1);
   }
-  if (translog_init(".", LOG_FILE_SIZE, 50112, 0, &pagecache, LOG_FLAGS))
+  if (translog_init_with_table(".", LOG_FILE_SIZE, 50112, 0, &pagecache,
+                               LOG_FLAGS, 0, &translog_example_table_init))
   {
     fprintf(stderr, "Can't init loghandler (%d)\n", errno);
-    translog_destroy();
     exit(1);
   }
-  example_loghandler_init();
   /* Suppressing of automatic record writing */
   dummy_transaction_object.first_undo_lsn|= TRANSACTION_LOGGED_LONG_ID;
 

storage/maria/unittest/ma_test_loghandler_nologs-t.c

+#include "../maria_def.h"
+#include <stdio.h>
+#include <errno.h>
+#include <tap.h>
+#include "../trnman.h"
+
+extern my_bool maria_log_remove();
+extern void example_loghandler_init();
+
+#ifndef DBUG_OFF
+static const char *default_dbug_option;
+#endif
+
+#define PCACHE_SIZE (1024*1024*10)
+#define PCACHE_PAGE TRANSLOG_PAGE_SIZE
+#define LOG_FILE_SIZE (8*1024L*1024L)
+#define LOG_FLAGS 0
+#define LONG_BUFFER_SIZE (LOG_FILE_SIZE + LOG_FILE_SIZE / 2)
+
+
+int main(int argc __attribute__((unused)), char *argv[])
+{
+  ulong i;
+  uint pagen;
+  uchar long_tr_id[6];
+  PAGECACHE pagecache;
+  LSN lsn;
+  LEX_STRING parts[TRANSLOG_INTERNAL_PARTS + 1];
+  uchar *long_buffer= malloc(LONG_BUFFER_SIZE);
+
+  MY_INIT(argv[0]);
+
+  plan(2);
+
+  bzero(&pagecache, sizeof(pagecache));
+  bzero(long_buffer, LONG_BUFFER_SIZE);
+  maria_data_root= ".";
+  if (maria_log_remove())
+    exit(1);
+
+  bzero(long_tr_id, 6);
+#ifndef DBUG_OFF
+#if defined(__WIN__)
+  default_dbug_option= "d:t:i:O,\\ma_test_loghandler.trace";
+#else
+  default_dbug_option= "d:t:i:o,/tmp/ma_test_loghandler.trace";
+#endif
+  if (argc > 1)
+  {
+    DBUG_SET(default_dbug_option);
+    DBUG_SET_INITIAL(default_dbug_option);
+  }
+#endif
+
+  if (ma_control_file_create_or_open(TRUE))
+  {
+    fprintf(stderr, "Can't init control file (%d)\n", errno);
+    exit(1);
+  }
+  if ((pagen= init_pagecache(&pagecache, PCACHE_SIZE, 0, 0,
+                             PCACHE_PAGE, 0)) == 0)
+  {
+    fprintf(stderr, "Got error: init_pagecache() (errno: %d)\n", errno);
+    exit(1);
+  }
+  if (translog_init_with_table(".", LOG_FILE_SIZE, 50112, 0, &pagecache,
+                               LOG_FLAGS, 0, &translog_example_table_init))
+  {
+    fprintf(stderr, "Can't init loghandler (%d)\n", errno);
+    exit(1);
+  }
+  /* Suppressing of automatic record writing */
+  dummy_transaction_object.first_undo_lsn|= TRANSACTION_LOGGED_LONG_ID;
+
+  /* write more then 1 file */
+  int4store(long_tr_id, 0);
+  parts[TRANSLOG_INTERNAL_PARTS + 0].str= (char*)long_tr_id;
+  parts[TRANSLOG_INTERNAL_PARTS + 0].length= 6;
+  if (translog_write_record(&lsn,
+                            LOGREC_FIXED_RECORD_0LSN_EXAMPLE,
+                            &dummy_transaction_object, NULL, 6,
+                            TRANSLOG_INTERNAL_PARTS + 1,
+                            parts, NULL, NULL))
+  {
+    fprintf(stderr, "Can't write record #%lu\n", (ulong) 0);
+    translog_destroy();
+    exit(1);
+  }
+
+  for(i= 0; i < LOG_FILE_SIZE/6 && LSN_FILE_NO(lsn) == 1; i++)
+  {
+    if (translog_write_record(&lsn,
+                              LOGREC_FIXED_RECORD_0LSN_EXAMPLE,
+                              &dummy_transaction_object, NULL, 6,
+                              TRANSLOG_INTERNAL_PARTS + 1,
+                              parts, NULL, NULL))
+    {
+      fprintf(stderr, "Can't write record #%lu\n", (ulong) 0);
+      translog_destroy();
+      exit(1);
+    }
+  }
+
+  translog_destroy();
+  end_pagecache(&pagecache, 1);
+  ma_control_file_end();
+
+  {
+    MY_STAT stat_buff;
+    char file_name[FN_REFLEN];
+    for (i= 1; i <= 2; i++)
+    {
+      translog_filename_by_fileno(i, file_name);
+      if (my_stat(file_name, &stat_buff, MY_WME) == NULL)
+      {
+        fprintf(stderr, "No file '%s'\n", file_name);
+        exit(1);
+      }
+      if (my_delete(file_name, MYF(MY_WME)) != 0)
+      {
+        fprintf(stderr, "Error %d during removing file'%s'\n",
+                errno, file_name);
+        exit(1);
+      }
+    }
+  }
+
+  if (ma_control_file_create_or_open(TRUE))
+  {
+    fprintf(stderr, "Can't init control file (%d)\n", errno);
+    exit(1);
+  }
+  if ((pagen= init_pagecache(&pagecache, PCACHE_SIZE, 0, 0,
+                             PCACHE_PAGE, 0)) == 0)
+  {
+    fprintf(stderr, "Got error: init_pagecache() (errno: %d)\n", errno);
+    exit(1);
+  }
+  if (translog_init_with_table(".", LOG_FILE_SIZE, 50112, 0, &pagecache,
+                               LOG_FLAGS, 0, &translog_example_table_init))
+  {
+    fprintf(stderr, "Can't init loghandler (%d)\n", errno);
+    exit(1);
+  }
+  /* Suppressing of automatic record writing */
+  dummy_transaction_object.first_undo_lsn|= TRANSACTION_LOGGED_LONG_ID;
+
+  ok(1, "Log init OK");
+
+  int4store(long_tr_id, 0);
+  parts[TRANSLOG_INTERNAL_PARTS + 0].str= (char*)long_tr_id;
+  parts[TRANSLOG_INTERNAL_PARTS + 0].length= 6;
+  if (translog_write_record(&lsn,
+                            LOGREC_FIXED_RECORD_0LSN_EXAMPLE,
+                            &dummy_transaction_object, NULL, 6,
+                            TRANSLOG_INTERNAL_PARTS + 1,
+                            parts, NULL, NULL))
+  {
+    fprintf(stderr, "Can't write record #%lu\n", (ulong) 0);
+    translog_destroy();
+    exit(1);
+  }
+
+  translog_destroy();
+  end_pagecache(&pagecache, 1);
+  ma_control_file_end();
+
+  if (!translog_is_file(3))
+  {
+    fprintf(stderr, "No file #3\n");
+    exit(1);
+  }
+
+  ok(1, "New log is OK");
+
+  if (maria_log_remove())
+    exit(1);
+  exit(0);
+}

storage/maria/unittest/ma_test_loghandler_pagecache-t.c

@@ -5,7 +5,7 @@
 #include "../trnman.h"
 
 extern my_bool maria_log_remove();
-extern void example_loghandler_init();
+extern void translog_example_table_init();
 
 #ifndef DBUG_OFF
 static const char *default_dbug_option;
@@ -20,6 +20,19 @@ static char *first_translog_file= (char*)"maria_log.00000001";
 static char *file1_name= (char*)"page_cache_test_file_1";
 static PAGECACHE_FILE file1;
 
+/**
+  @brief Dummy pagecache callback.
+*/
+
+static my_bool
+dummy_callback(__attribute__((unused)) uchar *page,
+               __attribute__((unused)) pgcache_page_no_t page_no,
+               __attribute__((unused)) uchar* data_ptr)
+{
+  return 0;
+}
+
+
 int main(int argc __attribute__((unused)), char *argv[])
 {
   uint pagen;
@@ -68,13 +81,12 @@ int main(int argc __attribute__((unused)), char *argv[])
     fprintf(stderr, "Got error: init_pagecache() (errno: %d)\n", errno);
     exit(1);
   }
-  if (translog_init(".", LOG_FILE_SIZE, 50112, 0, &pagecache, LOG_FLAGS))
+  if (translog_init_with_table(".", LOG_FILE_SIZE, 50112, 0, &pagecache,
+                               LOG_FLAGS, 0, &translog_example_table_init))
   {
     fprintf(stderr, "Can't init loghandler (%d)\n", errno);
-    translog_destroy();
     exit(1);
   }
-  example_loghandler_init();
   /* Suppressing of automatic record writing */
   dummy_transaction_object.first_undo_lsn|= TRANSACTION_LOGGED_LONG_ID;
 
@@ -112,6 +124,7 @@ int main(int argc __attribute__((unused)), char *argv[])
 	    errno);
     exit(1);
   }
+  pagecache_file_init(file1, &dummy_callback, &dummy_callback, NULL);
   if (chmod(file1_name, S_IRWXU | S_IRWXG | S_IRWXO) != 0)
   {
     fprintf(stderr, "Got error during file1 chmod() (errno: %d)\n",

storage/maria/unittest/ma_test_loghandler_purge-t.c

@@ -5,7 +5,7 @@
 #include "../trnman.h"
 
 extern my_bool maria_log_remove();
-extern void example_loghandler_init();
+extern void translog_example_table_init();
 
 #ifndef DBUG_OFF
 static const char *default_dbug_option;
@@ -63,13 +63,12 @@ int main(int argc __attribute__((unused)), char *argv[])
     fprintf(stderr, "Got error: init_pagecache() (errno: %d)\n", errno);
     exit(1);
   }
-  if (translog_init(".", LOG_FILE_SIZE, 50112, 0, &pagecache, LOG_FLAGS))
+  if (translog_init_with_table(".", LOG_FILE_SIZE, 50112, 0, &pagecache,
+                               LOG_FLAGS, 0, &translog_example_table_init))
   {
     fprintf(stderr, "Can't init loghandler (%d)\n", errno);
-    translog_destroy();
     exit(1);
   }
-  example_loghandler_init();
   /* Suppressing of automatic record writing */
   dummy_transaction_object.first_undo_lsn|= TRANSACTION_LOGGED_LONG_ID;
 

storage/maria/unittest/test_file.c

@@ -30,11 +30,6 @@ int test_file(PAGECACHE_FILE file, char *file_name,
   int step= 0;
   int res= 1;                                   /* ok */
 
-  if (my_sync(file.file, MYF(MY_WME | MY_IGNORE_BADFD)))
-  {
-    diag("Got error during syncing file\n");
-    exit(1);
-  }
   if ((stat= my_stat(file_name, &stat_buff, MYF(0))) == NULL)
   {
     diag("Can't stat() %s (errno: %d)\n", file_name, errno);