- WL#3239 "log CREATE TABLE in Maria"

- WL#3240 "log DROP TABLE in Maria"
- similarly, log RENAME TABLE, REPAIR/OPTIMIZE TABLE, and
DELETE no_WHERE_clause (== the DELETE which just truncates the files)
- create_rename_lsn added to MARIA_SHARE's state
- all these operations (except DROP TABLE) also update the table's
create_rename_lsn, which is needed for the correctness of
Recovery (see function comment of _ma_repair_write_log_record()
in ma_check.c)
- write a COMMIT record when transaction commits.
- don't log REDOs/UNDOs if this is an internal temporary table
like inside ALTER TABLE (I expect this to be a big win). There was
already no logging for user-created "CREATE TEMPORARY" tables.
- don't fsync files/directories if the table is not transactional
- in translog_write_record(), autogenerate a 2-byte-id for the table
and log the "id->name" pair (LOGREC_FILE_ID); log
LOGREC_LONG_TRANSACTION_ID; automatically store
the table's 2-byte-id in any log record.
- preparations for Checkpoint: translog_get_horizon(); pausing Checkpoint
when some dirty pages are unknown; capturing trn->rec_lsn,
trn->first_undo_lsn for Checkpoint and log's low-water-mark computing.
- assertions, comments.


storage/maria/Makefile.am:
  more files to build
storage/maria/ha_maria.cc:
  - logging a REPAIR log record if REPAIR/OPTIMIZE was successful.
  - ha_maria::data_file_type does not have to be set in every info()
  call, just do it once in open().
  - if caller said that transactionality can be disabled (like if
  caller is ALTER TABLE) i.e. thd->transaction.on==FALSE, then we
  temporarily disable transactionality of the table in external_lock();
  that will ensure that no REDOs/UNDOs are logged for this possibly
  massive write operation (they are not needed, as if any write fails,
  the table will be dropped). We re-enable in external_lock(F_UNLCK),
  which in ALTER TABLE happens before the tmp table replaces the original
  one (which is good, as thus the final table will have a REDO RENAME
  and a correct create_rename_lsn).
  - when we commit we also have to write a log record, so
  trnman_commit_trn() calls become ma_commit() calls
  - at end of engine's initialization, we are potentially entering a
  multi-threaded dangerous world (clients are going to be accepted)
  and so some assertions of mutex-owning become enforceable, for that
  we set maria_multi_threaded=TRUE (see ma_control_file.c)
storage/maria/ha_maria.h:
  new member ha_maria::save_transactional (see also ha_maria.cc)
storage/maria/ma_blockrec.c:
  - fixing comments according to discussion with Monty
  - if a table is transactional but temporarily non-transactional
  (like in ALTER TABLE), we need to give a sensible LSN to the pages
  (and, if we give 0, pagecache asserts).
  - translog_write_record() now takes care of storing the share's
  2-byte-id in the log record
storage/maria/ma_blockrec.h:
  fixing comment according to discussion with Monty
storage/maria/ma_check.c:
  When REPAIR/OPTIMIZE modify the data/index file, if this is a
  transactional table, they must sync it; if they remove files or rename
  files, they must sync the directory, so that everything is durable.
  This is just applying to REPAIR/OPTIMIZE the logic already implemented
  in CREATE/DROP/RENAME a few months ago.
  Adding a function to write a LOGREC_REPAIR_TABLE at end of
  REPAIR/OPTIMIZE (called only by ha_maria, not by maria_chk), and
  to update the table's create_rename_lsn.
storage/maria/ma_close.c:
  fix for a future bug
storage/maria/ma_control_file.c:
  ensuring that if Maria is running in multi-threaded mode, anybody
  wanting to write to the control file and update
  last_checkpoint_lsn/last_logno owns the log's lock.
storage/maria/ma_control_file.h:
  see ma_control_file.c
storage/maria/ma_create.c:
  when creating a table:
  - sync it and its directory only if this is a transactional table
  and there is a log (no point in syncing in maria_chk)
  - decouple the two uses of linkname/linkname_ptr (for index file and
  for data file) into more variables, as we need to know all links
  until the moment we write the LOGREC_CREATE_TABLE.
  - set share.data_file_type early so that _ma_initialize_data_file()
  knows it (Monty's bugfix so that a table always has at least a bitmap
  page when it is created; so data-file is not 0 bytes anymore).
  - log a LOGREC_CREATE_TABLE; it contains the bytes which we have
  just written to the index file's header. Update table's
  create_rename_lsn.
  - syncing of kfile had been bugified in a previous merge, correcting
  - syncing of dfile is now needed as it's not empty anymore
  - in _ma_initialize_data_file(), use share's block_size and not the
  global one. This is a gratuitous change, both variables are equal,
  just that I find it more future-proof to use share-bound variable
  rather than global one.
storage/maria/ma_delete_all.c:
  log a LOGREC_DELETE_ALL record when doing ma_delete_all_rows();
  update create_rename_lsn then.
storage/maria/ma_delete_table.c:
  - logging LOGREC_DROP_TABLE; knowing if this is needed, requires
  knowing if the table is transactional, which requires opening the
  table.
  - we need to sync directories only if the table is transactional
storage/maria/ma_extra.c:
  questions
storage/maria/ma_init.c:
  when maria_end() is called, engine is not multithreaded
storage/maria/ma_loghandler.c:
  - translog_inited has to be visible to ma_create() (see how it is used
  in ma_create())
  - checkpoint record will be a single record, not three
  - no REDO for TRUNCATE (TRUNCATE calls ma_create() internally so will
  log a REDO_CREATE)
  - adding REDO for DELETE no_WHERE_clause (fast DELETE of all rows by
  truncating the files), REPAIR.
  - MY_WAIT_IF_FULL to wait&retry if a log write hits a full disk
  - in translog_write_record(), if MARIA_SHARE does not yet have a
  2-byte-id, generate one for it and log LOGREC_FILE_ID; automatically
  store this short id into log records.
  - in translog_write_record(), if transaction has not logged its
  long trid, log LOGREC_LONG_TRANSACTION_ID.
  - For Checkpoint, we need to know the current end-of-log: adding
  translog_get_horizon().
  - For Control File, adding an assertion that the thread owns the
  log's lock (control file is protected by this lock)
storage/maria/ma_loghandler.h:
  Changes in log records (see ma_loghandler.c).
  new prototypes, new functions.
storage/maria/ma_loghandler_lsn.h:
  adding a type LSN_WITH_FLAGS especially for TRN::first_undo_lsn,
  where the most significant byte is used for flags.
storage/maria/ma_open.c:
  storing the create_rename_lsn in the index file's header (in the
  state, precisely) and retrieving it from there.
storage/maria/ma_pagecache.c:
  - my set_if_bigger was wrong, correcting it
  - if the first_in_switch list is not empty, it means that
  changed_blocks misses some dirty pages, so Checkpoint cannot run and
  needs to wait. A variable missing_blocks_in_changed_list is added to
  tell that (should it be named missing_blocks_in_changed_blocks?)
  - pagecache_collect_changed_blocks_with_lsn() now also tells the
  minimum rec_lsn (needed for low-water mark computation).
storage/maria/ma_pagecache.h:
  see ma_pagecache.c
storage/maria/ma_panic.c:
  comment
storage/maria/ma_range.c:
  comment
storage/maria/ma_rename.c:
  - logging LOGREC_RENAME_TABLE; knowing if this is needed, requires
  knowing if the table is transactional, which requires opening the
  table.
  - update create_rename_lsn
  - we need to sync directories only if the table is transactional
storage/maria/ma_static.c:
  comment
storage/maria/ma_test_all.sh:
  - tip for Valgrind-ing ma_test_all
  - do "export maria_path=somepath" before calling ma_test_all,
  if you want to run ma_test_all out of storage/maria (useful
  to have parallel runs, like one normal and one Valgrind, they
  must not use the same tables so need to run in different directories)
storage/maria/maria_def.h:
  - state now contains, in memory and on disk, the create_rename_lsn
  - share now contains a 2-byte-id
storage/maria/trnman.c:
  preparations for Checkpoint: capture trn->rec_lsn, trn->first_undo_lsn;
  minimum first_undo_lsn needed to know log's low-water-mark
storage/maria/trnman.h:
  using most significant byte of first_undo_lsn to hold miscellaneous
  flags, for now TRANSACTION_LOGGED_LONG_ID.
  dummy_transaction_object is already declared in ma_static.c.
storage/maria/trnman_public.h:
  dummy_transaction_object was declared in all files including
  trnman_public.h, while in fact it's a single object.
  new prototype
storage/maria/unittest/ma_test_loghandler-t.c:
  update for new prototype
storage/maria/unittest/ma_test_loghandler_multigroup-t.c:
  update for new prototype
storage/maria/unittest/ma_test_loghandler_multithread-t.c:
  update for new prototype
storage/maria/unittest/ma_test_loghandler_pagecache-t.c:
  update for new prototype
storage/maria/ma_commit.c:
  function which wraps:
  - writing a LOGREC_COMMIT record (==commit on disk)
  - calling trnman_commit_trn() (=commit in memory)
storage/maria/ma_commit.h:
  new header file
.tree-is-private:
  this file is now needed to keep our tree private (don't push it
  to public trees). When 5.1 is merged into mysql-maria, we can abandon
  our maria-specific post-commit trigger; .tree_is_private will take
  care of keeping commit mails private. Don't push this file to public
  trees.

storage/maria/Makefile.am

@@ -54,7 +54,8 @@ noinst_HEADERS =	maria_def.h ma_rt_index.h ma_rt_key.h ma_rt_mbr.h \
 			ma_sp_defs.h ma_fulltext.h ma_ftdefs.h ma_ft_test1.h \
 			ma_ft_eval.h trnman.h lockman.h tablockman.h \
 			ma_control_file.h ha_maria.h ma_blockrec.h \
-			ma_loghandler.h ma_loghandler_lsn.h ma_pagecache.h
+			ma_loghandler.h ma_loghandler_lsn.h ma_pagecache.h \
+			ma_commit.h
 ma_test1_DEPENDENCIES=	$(LIBRARIES)
 ma_test1_LDADD=		@CLIENT_EXTRA_LDFLAGS@ libmaria.a \
                         $(top_builddir)/storage/myisam/libmyisam.a \
@@ -112,7 +113,8 @@ libmaria_a_SOURCES =	ma_init.c ma_open.c ma_extra.c ma_info.c ma_rkey.c \
 			ha_maria.cc trnman.c lockman.c tablockman.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_pagecache.c ma_pagecaches.c \
+			ma_commit.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

@@ -30,6 +30,7 @@
 #include "maria_def.h"
 #include "ma_rt_index.h"
 #include "ma_blockrec.h"
+#include "ma_commit.h"
 
 #define MARIA_CANNOT_ROLLBACK HA_NO_TRANSACTIONS
 #ifdef MARIA_CANNOT_ROLLBACK
@@ -690,7 +691,8 @@ int ha_maria::open(const char *name, int mode, uint test_if_locked)
   info(HA_STATUS_NO_LOCK | HA_STATUS_VARIABLE | HA_STATUS_CONST);
   if (!(test_if_locked & HA_OPEN_WAIT_IF_LOCKED))
     VOID(maria_extra(file, HA_EXTRA_WAIT_LOCK, 0));
-  if (file->s->data_file_type != STATIC_RECORD)
+  save_transactional= file->s->base.transactional;
+  if ((data_file_type= file->s->data_file_type) != STATIC_RECORD)
     int_table_flags |= HA_REC_NOT_IN_SEQ;
   if (file->s->options & (HA_OPTION_CHECKSUM | HA_OPTION_COMPRESS_RECORD))
     int_table_flags |= HA_HAS_CHECKSUM;
@@ -1178,6 +1180,8 @@ int ha_maria::repair(THD *thd, HA_CHECK &param, bool do_optimize)
                               llstr(rows, llbuff),
                               llstr(file->state->records, llbuff2));
     }
+    if (!error)
+      error= _ma_repair_write_log_record(&param, file);
   }
   else
   {
@@ -1806,7 +1810,6 @@ int ha_maria::info(uint flag)
               MY_APPEND_EXT | MY_UNPACK_FILENAME);
     if (strcmp(name_buff, maria_info.index_file_name))
       index_file_name=maria_info.index_file_name;
-    data_file_type= maria_info.data_file_type;
   }
   if (flag & HA_STATUS_ERRKEY)
   {
@@ -1860,7 +1863,7 @@ int ha_maria::external_lock(THD *thd, int lock_type)
 {
   TRN *trn= THD_TRN;
   DBUG_ENTER("ha_maria::external_lock");
-  if (!file->s->base.transactional)
+  if (!save_transactional)
     goto skip_transaction;
   if (!trn && lock_type != F_UNLCK) /* no transaction yet - open it now */
   {
@@ -1884,6 +1887,19 @@ int ha_maria::external_lock(THD *thd, int lock_type)
       trans_register_ha(thd, FALSE, maria_hton);
       trnman_new_statement(trn);
     }
+    if (!thd->transaction.on)
+    {
+      /*
+        No need to log REDOs/UNDOs. If this is an internal temporary table
+        which will be renamed to a permanent table (like in ALTER TABLE),
+        the rename happens after unlocking so will be durable (and the table
+        will get its create_rename_lsn).
+        Note: if we wanted to enable users to have an old backup and apply
+        tons of archived logs to roll-forward, we could then not disable
+        REDOs/UNDOs in this case.
+      */
+      file->s->base.transactional= FALSE;
+    }
   }
   else
   {
@@ -1894,7 +1910,8 @@ int ha_maria::external_lock(THD *thd, int lock_type)
       {
         /* autocommit ? rollback a transaction */
 #ifdef MARIA_CANNOT_ROLLBACK
-        trnman_commit_trn(trn);
+        if (ma_commit(trn))
+          DBUG_RETURN(1);
         THD_TRN= 0;
 #else
         if (!(thd->options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN)))
@@ -1906,6 +1923,7 @@ int ha_maria::external_lock(THD *thd, int lock_type)
 #endif
       }
     }
+    file->s->base.transactional= save_transactional;
   }
 skip_transaction:
   DBUG_RETURN(maria_lock_database(file, !table->s->tmp_table ?
@@ -1916,7 +1934,7 @@ skip_transaction:
 int ha_maria::start_stmt(THD *thd, thr_lock_type lock_type)
 {
   TRN *trn= THD_TRN;
-  if (file->s->base.transactional)
+  if (save_transactional)
   {
     DBUG_ASSERT(trn); // this may be called only after external_lock()
     DBUG_ASSERT(trnman_has_locked_tables(trn));
@@ -2186,8 +2204,7 @@ static int maria_commit(handlerton *hton __attribute__ ((unused)),
     DBUG_RETURN(0); // end of statement
   DBUG_PRINT("info", ("THD_TRN set to 0x0"));
   THD_TRN= 0;
-  DBUG_RETURN(trnman_commit_trn(trn) ?
-              HA_ERR_OUT_OF_MEM : 0); // end of transaction
+  DBUG_RETURN(ma_commit(trn)); // end of transaction
 }
 
 
@@ -2212,6 +2229,7 @@ static int maria_rollback(handlerton *hton __attribute__ ((unused)),
 
 static int ha_maria_init(void *p)
 {
+  int res;
   maria_hton= (handlerton *)p;
   maria_hton->state= SHOW_OPTION_YES;
   maria_hton->db_type= DB_TYPE_MARIA;
@@ -2223,14 +2241,16 @@ static int ha_maria_init(void *p)
   maria_hton->flags= HTON_CAN_RECREATE | HTON_SUPPORT_LOG_TABLES;
   bzero(maria_log_pagecache, sizeof(*maria_log_pagecache));
   maria_data_root= mysql_real_data_home;
-  return (test(maria_init() || ma_control_file_create_or_open() ||
-               (init_pagecache(maria_log_pagecache,
-                               TRANSLOG_PAGECACHE_SIZE, 0, 0,
-                               TRANSLOG_PAGE_SIZE) == 0) ||
-               translog_init(maria_data_root, TRANSLOG_FILE_SIZE,
-                             MYSQL_VERSION_ID, server_id, maria_log_pagecache,
-                             TRANSLOG_DEFAULT_FLAGS) ||
-               trnman_init()));
+  res= maria_init() || ma_control_file_create_or_open() ||
+    (init_pagecache(maria_log_pagecache,
+                    TRANSLOG_PAGECACHE_SIZE, 0, 0,
+                    TRANSLOG_PAGE_SIZE) == 0) ||
+    translog_init(maria_data_root, TRANSLOG_FILE_SIZE,
+                  MYSQL_VERSION_ID, server_id, maria_log_pagecache,
+                  TRANSLOG_DEFAULT_FLAGS) ||
+    trnman_init();
+  maria_multi_threaded= TRUE;
+  return res;
 }
 
 

storage/maria/ha_maria.h

@@ -39,6 +39,11 @@ class ha_maria :public handler
   char *data_file_name, *index_file_name;
   enum data_file_type data_file_type;
   bool can_enable_indexes;
+  /**
+     @brief for temporarily disabling table's transactionality
+     (if THD::transaction::on is false), remember the original value here
+  */
+  bool save_transactional;
   int repair(THD * thd, HA_CHECK &param, bool optimize);
 
 public:

storage/maria/ma_blockrec.h

@@ -96,7 +96,7 @@ enum en_page_type { UNALLOCATED_PAGE, HEAD_PAGE, TAIL_PAGE, BLOB_PAGE, MAX_PAGE_
 /******* defines that affects allocation (density) of data *******/
 
 /*
-  If the tail part (from the main block or a blob) uses more than 75 % of
+  If the tail part (from the main block or a blob) would use more than 75 % of
   the size of page, store the tail on a full page instead of a shared
  tail page.
 */

storage/maria/ma_check.c

@@ -53,6 +53,7 @@
 #endif
 #include "ma_rt_index.h"
 #include "ma_blockrec.h"
+#include "trnman_public.h"
 
 /* Functions defined in this file */
 
@@ -2132,11 +2133,15 @@ err:
     /* Replace the actual file with the temporary file */
     if (new_file >= 0)
     {
+      myf sync_dir= (share->base.transactional && !share->temporary) ?
+        MY_SYNC_DIR : 0;
       my_close(new_file,MYF(0));
       info->dfile.file= new_file= -1;
       if (maria_change_to_newfile(share->data_file_name,MARIA_NAME_DEXT,
-			    DATA_TMP_EXT, (param->testflag & T_BACKUP_DATA ?
-			     MYF(MY_REDEL_MAKE_BACKUP): MYF(0))) ||
+                                  DATA_TMP_EXT,
+                                  MYF((param->testflag & T_BACKUP_DATA ?
+                                       MY_REDEL_MAKE_BACKUP : 0) |
+                                      sync_dir)) ||
 	  _ma_open_datafile(info,share,-1))
 	got_error=1;
     }
@@ -2328,6 +2333,8 @@ int maria_sort_index(HA_CHECK *param, register MARIA_HA *info, my_string name)
   int old_lock;
   MARIA_SHARE *share=info->s;
   MARIA_STATE_INFO old_state;
+  myf sync_dir= (share->base.transactional && !share->temporary) ?
+    MY_SYNC_DIR : 0;
   DBUG_ENTER("maria_sort_index");
 
   /* cannot sort index files with R-tree indexes */
@@ -2388,7 +2395,7 @@ int maria_sort_index(HA_CHECK *param, register MARIA_HA *info, my_string name)
   share->kfile.file = -1;
   VOID(my_close(new_file,MYF(MY_WME)));
   if (maria_change_to_newfile(share->index_file_name, MARIA_NAME_IEXT,
-                              INDEX_TMP_EXT, MYF(0)) ||
+                              INDEX_TMP_EXT, sync_dir) ||
       _ma_open_keyfile(share))
     goto err2;
   info->lock_type= F_UNLCK;			/* Force maria_readinfo to lock */
@@ -2604,6 +2611,8 @@ int maria_repair_by_sort(HA_CHECK *param, register MARIA_HA *info,
   char llbuff[22];
   MARIA_SORT_INFO sort_info;
   ulonglong key_map=share->state.key_map;
+  myf sync_dir= (share->base.transactional && !share->temporary) ?
+    MY_SYNC_DIR : 0;
   DBUG_ENTER("maria_repair_by_sort");
 
   start_records=info->state->records;
@@ -2922,8 +2931,9 @@ err:
       info->dfile.file= new_file= -1;
       if (maria_change_to_newfile(share->data_file_name,MARIA_NAME_DEXT,
                                   DATA_TMP_EXT,
-                                  (param->testflag & T_BACKUP_DATA ?
-                                   MYF(MY_REDEL_MAKE_BACKUP): MYF(0))) ||
+                                  MYF((param->testflag & T_BACKUP_DATA ?
+                                       MY_REDEL_MAKE_BACKUP : 0) |
+                                      sync_dir)) ||
 	  _ma_open_datafile(info,share,-1))
 	got_error=1;
     }
@@ -3022,6 +3032,8 @@ int maria_repair_parallel(HA_CHECK *param, register MARIA_HA *info,
   MARIA_SORT_INFO sort_info;
   ulonglong key_map=share->state.key_map;
   pthread_attr_t thr_attr;
+  myf sync_dir= (share->base.transactional && !share->temporary) ?
+    MY_SYNC_DIR : 0;
   DBUG_ENTER("maria_repair_parallel");
 
   start_records=info->state->records;
@@ -3445,8 +3457,9 @@ err:
       info->dfile.file= new_file= -1;
       if (maria_change_to_newfile(share->data_file_name,MARIA_NAME_DEXT,
                                   DATA_TMP_EXT,
-                                  (param->testflag & T_BACKUP_DATA ?
-                                   MYF(MY_REDEL_MAKE_BACKUP): MYF(0))) ||
+                                  MYF((param->testflag & T_BACKUP_DATA ?
+                                       MY_REDEL_MAKE_BACKUP : 0) |
+                                      sync_dir)) ||
 	  _ma_open_datafile(info,share,-1))
 	got_error=1;
     }
@@ -5135,3 +5148,64 @@ static void restore_data_file_type(MARIA_SHARE *share)
   share->data_file_type= share->state.header.data_file_type=
   share->pack.header_length= 0;
 }
+
+
+/**
+   @brief Writes a LOGREC_REPAIR_TABLE record and updates create_rename_lsn
+
+   REPAIR/OPTIMIZE have replaced the data/index file with a new file
+   and so, in this scenario:
+   @verbatim
+     CHECKPOINT - REDO_INSERT - COMMIT - ... - REPAIR - ... - crash
+   @endverbatim
+   we do not want Recovery to apply the REDO_INSERT to the table, as it would
+   then possibly wrongly extend the table. By updating create_rename_lsn at
+   the end of REPAIR, we know that REDO_INSERT will be skipped.
+
+   @param  param            description of the REPAIR operation
+   @param  info             table
+
+   @return Operation status
+     @retval 0      ok
+     @retval 1      error (disk problem)
+*/
+
+int _ma_repair_write_log_record(const HA_CHECK *param, MARIA_HA *info)
+{
+  MARIA_SHARE *share= info->s;
+  /* Only called from ha_maria.cc, not maria_check, so translog is inited */
+  if (share->base.transactional && !share->temporary)
+  {
+    /* For now this record is only informative */
+    LEX_STRING log_array[TRANSLOG_INTERNAL_PARTS + 1];
+    uchar log_data[LSN_STORE_SIZE];
+    compile_time_assert(LSN_STORE_SIZE >= (FILEID_STORE_SIZE + 4));
+    log_array[TRANSLOG_INTERNAL_PARTS + 0].str=    (char*) log_data;
+    log_array[TRANSLOG_INTERNAL_PARTS + 0].length= FILEID_STORE_SIZE + 4;
+    /*
+      testflag gives an idea of what REPAIR did (in particular T_QUICK
+      or not: did it touch the data file or not?).
+    */
+    int4store(log_data + FILEID_STORE_SIZE, param->testflag);
+    if (unlikely(translog_write_record(&share->state.create_rename_lsn,
+                                       LOGREC_REDO_REPAIR_TABLE,
+                                       &dummy_transaction_object, share,
+                                       log_array[TRANSLOG_INTERNAL_PARTS +
+                                                 0].length,
+                                       sizeof(log_array)/sizeof(log_array[0]),
+                                       log_array, log_data)))
+      return 1;
+    /*
+      But this piece is really needed, to have the new table's content durable
+      and to not apply old REDOs to the new table. The table's existence was
+      made durable earlier (MY_SYNC_DIR passed to maria_change_to_newfile()).
+    */
+    lsn_store(log_data, share->state.create_rename_lsn);
+    DBUG_ASSERT(info->dfile.file >= 0);
+    DBUG_ASSERT(share->kfile.file >= 0);
+    return (my_pwrite(share->kfile.file, log_data, sizeof(log_data),
+                      sizeof(share->state.header) + 2, MYF(MY_NABP)) ||
+            _ma_sync_table_files(info));
+  }
+  return 0;
+}

storage/maria/ma_close.c

@@ -57,14 +57,6 @@ int maria_close(register MARIA_HA *info)
     info->opt_flag&= ~(READ_CACHE_USED | WRITE_CACHE_USED);
   }
   flag= !--share->reopen;
-  /*
-    RECOVERY TODO:
-    If "flag" is TRUE, in the line below we are going to make the table
-    unknown to future checkpoints, so it needs to have fsync'ed itself
-    entirely (bitmap, pages, etc) at this point.
-    The flushing is currently done a few lines further (which is ok, as we
-    still hold THR_LOCK_maria), but syncing is missing.
-  */
   maria_open_list=list_delete(maria_open_list,&info->open_list);
   pthread_mutex_unlock(&share->intern_lock);
 
@@ -82,7 +74,12 @@ int maria_close(register MARIA_HA *info)
                                   FLUSH_IGNORE_CHANGED :
                                   FLUSH_RELEASE)))
         error= my_errno;
-
+      /*
+        File must be synced as it is going out of the maria_open_list and so
+        becoming unknown to Checkpoint.
+      */
+      if (my_sync(share->kfile.file, MYF(MY_WME)))
+        error= my_errno;
       /*
         If we are crashed, we can safely flush the current state as it will
         not change the crashed state.

storage/maria/ma_commit.c

+/* Copyright (C) 2007 MySQL AB
+
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; version 2 of the License.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
+
+#include "maria_def.h"
+#include "trnman.h"
+
+/**
+   @brief writes a COMMIT record to log and commits transaction in memory
+
+   @param  trn              transaction
+
+   @return Operation status
+     @retval 0      ok
+     @retval 1      error (disk error or out of memory)
+*/
+
+int ma_commit(TRN *trn)
+{
+  if (trn->undo_lsn == 0) /* no work done, rollback (cheaper than commit) */
+    return trnman_rollback_trn(trn);
+  /*
+    - if COMMIT record is written before trnman_commit_trn():
+    if Checkpoint comes in the middle it will see trn is not committed,
+    then if crash, Recovery might roll back trn (if min(rec_lsn) is after
+    COMMIT record) and this is not an issue as
+    * transaction's updates were not made visible to other transactions
+    * "commit ok" was not sent to client
+    Alternatively, Recovery might commit trn (if min(rec_lsn) is before COMMIT
+    record), which is ok too. All in all it means that "trn committed" is not
+    100% equal to "COMMIT record written".
+    - if COMMIT record is written after trnman_commit_trn():
+    if crash happens between the two, trn will be rolled back which is an
+    issue (transaction's updates were made visible to other transactions).
+    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.
+  */
+  LSN commit_lsn;
+  LEX_STRING log_array[TRANSLOG_INTERNAL_PARTS];
+  /*
+    We do not store "thd->transaction.xid_state.xid" for now, it will be
+    needed only when we support XA.
+  */
+  return
+    translog_write_record(&commit_lsn, LOGREC_COMMIT,
+                          trn, NULL, 0,
+                          sizeof(log_array)/sizeof(log_array[0]),
+                          log_array, NULL) ||
+    translog_flush(commit_lsn) || trnman_commit_trn(trn);
+  /*
+    Note: if trnman_commit_trn() fails above, we have already
+    written the COMMIT record, so Checkpoint and Recovery will see the
+    transaction as committed.
+  */
+}

storage/maria/ma_commit.h

+/* Copyright (C) 2007 MySQL AB
+
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; version 2 of the License.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
+
+C_MODE_START
+int ma_commit(TRN *trn);
+C_MODE_END

storage/maria/ma_control_file.c

@@ -50,6 +50,13 @@
 LSN last_checkpoint_lsn;
 uint32 last_logno;
 
+/**
+   @brief If log's lock should be asserted when writing to control file.
+
+   Can be re-used by any function which needs to be thread-safe except when
+   it is called at startup.
+*/
+my_bool maria_multi_threaded= FALSE;
 
 /*
   Control file is less then  512 bytes (a disk sector),
@@ -203,6 +210,8 @@ err:
   the last_checkpoint_lsn and last_logno global variables.
   Called when we have created a new log (after syncing this log's creation)
   and when we have written a checkpoint (after syncing this log record).
+  Variables last_checkpoint_lsn and last_logno must be protected by caller
+  using log's lock, unless this function is called at startup.
 
   SYNOPSIS
     ma_control_file_write_and_force()
@@ -233,12 +242,14 @@ int ma_control_file_write_and_force(const LSN checkpoint_lsn, uint32 logno,
   DBUG_ENTER("ma_control_file_write_and_force");
 
   DBUG_ASSERT(control_file_fd >= 0); /* must be open */
+#ifndef DBUG_OFF
+  if (maria_multi_threaded)
+    translog_lock_assert_owner();
+#endif
 
   memcpy(buffer + CONTROL_FILE_MAGIC_STRING_OFFSET,
          CONTROL_FILE_MAGIC_STRING, CONTROL_FILE_MAGIC_STRING_SIZE);
 
-  /* TODO: you need some protection to be able to read last_* global vars */
-
   if (objs_to_write == CONTROL_FILE_UPDATE_ONLY_LSN)
     update_checkpoint_lsn= TRUE;
   else if (objs_to_write == CONTROL_FILE_UPDATE_ONLY_LOGNO)
@@ -270,7 +281,6 @@ int ma_control_file_write_and_force(const LSN checkpoint_lsn, uint32 logno,
       my_sync(control_file_fd, MYF(MY_WME)))
     DBUG_RETURN(1);
 
-  /* TODO: you need some protection to be able to write last_* global vars */
   if (update_checkpoint_lsn)
     last_checkpoint_lsn= checkpoint_lsn;
   if (update_logno)

storage/maria/ma_control_file.h

@@ -43,6 +43,8 @@ extern LSN last_checkpoint_lsn;
 */
 extern uint32 last_logno;
 
+extern my_bool maria_multi_threaded;
+
 typedef enum enum_control_file_error {
   CONTROL_FILE_OK= 0,
   CONTROL_FILE_TOO_SMALL,

storage/maria/ma_delete_all.c

@@ -17,21 +17,38 @@
 /* This clears the status information and truncates files */
 
 #include "maria_def.h"
+#include "trnman_public.h"
+
+/**
+   @brief deletes all rows from a table
+
+   @param  info             Maria handler
+
+   @return Operation status
+     @retval 0      ok
+     @retval 1      error
+*/
 
 int maria_delete_all_rows(MARIA_HA *info)
 {
   uint i;
   MARIA_SHARE *share=info->s;
   MARIA_STATE_INFO *state=&share->state;
+  my_bool log_record;
   DBUG_ENTER("maria_delete_all_rows");
 
   if (share->options & HA_OPTION_READ_ONLY_DATA)
   {
     DBUG_RETURN(my_errno=EACCES);
   }
-  /* LOCK TODO take X-lock on table here */
+  /**
+     @todo LOCK take X-lock on table here.
+     When we have versioning, if some other thread is looking at this table,
+     we cannot shrink the file like this.
+  */
   if (_ma_readinfo(info,F_WRLCK,1))
     DBUG_RETURN(my_errno);
+  log_record= share->base.transactional && !share->temporary;
   if (_ma_mark_file_changed(info))
     goto err;
 
@@ -54,27 +71,13 @@ int maria_delete_all_rows(MARIA_HA *info)
   */
   flush_pagecache_blocks(share->pagecache, &share->kfile,
                          FLUSH_IGNORE_CHANGED);
-  /*
-    RECOVERY TODO Log the two chsize and header modifications and force the
-    log. So that if crash between the two chsize, we finish the work at
-    Recovery. For this scenario:
-    "TRUNCATE TABLE t1; DROP TABLE t1; RENAME TABLE t2 to t1; crash;"
-    Recovery mustn't truncate the new t1, so the log records of TRUNCATE
-    should be applied only if t1 exists and its ZeroDirtyPagesLSN is smaller
-    than the records'. See more comments below.
-  */
   if (my_chsize(info->dfile.file, 0, 0, MYF(MY_WME)) ||
       my_chsize(share->kfile.file, share->base.keystart, 0, MYF(MY_WME))  )
     goto err;
 
-  if (_ma_initialize_data_file(info->dfile.file, info->s))
+  if (_ma_initialize_data_file(info->dfile.file, share))
     goto err;
 
-  /*
-    RECOVERY TODO Consider updating ZeroDirtyPagesLSN here. It is
-    not a necessity (it is one only in RENAME commands) but an optional
-    optimization which will allow some REDO skipping at Recovery.
-  */
   VOID(_ma_writeinfo(info,WRITEINFO_UPDATE_KEYFILE));
 #ifdef HAVE_MMAP
   /* Resize mmaped area */
@@ -82,24 +85,48 @@ int maria_delete_all_rows(MARIA_HA *info)
   _ma_remap_file(info, (my_off_t)0);
   rw_unlock(&info->s->mmap_lock);
 #endif
-  /*
-    RECOVERY TODO Until we have the TRUNCATE log record and take it into
-    account for log-low-water-mark calculation and use it in Recovery, we need
-    to sync.
-  */
-  if (_ma_sync_table_files(info))
-    goto err;
+  if (log_record)
+  {
+    /* For now this record is only informative */
+    LEX_STRING log_array[TRANSLOG_INTERNAL_PARTS + 1];
+    uchar log_data[LSN_STORE_SIZE];
+    log_array[TRANSLOG_INTERNAL_PARTS + 0].str=    (char*) log_data;
+    log_array[TRANSLOG_INTERNAL_PARTS + 0].length= FILEID_STORE_SIZE;
+    if (unlikely(translog_write_record(&share->state.create_rename_lsn,
+                                       LOGREC_REDO_DELETE_ALL,
+                                       info->trn, share, 0,
+                                       sizeof(log_array)/sizeof(log_array[0]),
+                                       log_array, log_data)))
+      goto err;
+    /*
+      store LSN into file. It is an optimization so that all old REDOs for
+      this table are ignored (scenario: checkpoint, INSERT1s, DELETE ALL;
+      INSERT2s, crash: then Recovery can skip INSERT1s). It also allows us to
+      ignore the present record at Recovery.
+      Note that storing the LSN could not be done by _ma_writeinfo() above as
+      the table is locked at this moment. So we need to do it by ourselves.
+    */
+    lsn_store(log_data, share->state.create_rename_lsn);
+    if (my_pwrite(share->kfile.file, log_data, sizeof(log_data),
+                  sizeof(share->state.header) + 2, MYF(MY_NABP)) ||
+        _ma_sync_table_files(info))
+      goto err;
+    /**
+       @todo RECOVERY Until we take into account the log record above
+       for log-low-water-mark calculation and use it in Recovery, we need
+       to sync above.
+    */
+  }
   allow_break();			/* Allow SIGHUP & SIGINT */
   DBUG_RETURN(0);
 
 err:
   {
     int save_errno=my_errno;
-    /* RECOVERY TODO log the header modifications */
     VOID(_ma_writeinfo(info,WRITEINFO_UPDATE_KEYFILE));
     info->update|=HA_STATE_WRITTEN;	/* Buffer changed */
-    /* RECOVERY TODO until we log above we have to sync */
-    if (_ma_sync_table_files(info) && !save_errno)
+    /** @todo RECOVERY until we use the log record above we have to sync */
+    if (log_record &&_ma_sync_table_files(info) && !save_errno)
       save_errno= my_errno;
     allow_break();			/* Allow SIGHUP & SIGINT */
     DBUG_RETURN(my_errno=save_errno);

storage/maria/ma_delete_table.c

@@ -13,11 +13,18 @@
    along with this program; if not, write to the Free Software
    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
 
-/*
-  deletes a table
-*/
-
 #include "ma_fulltext.h"
+#include "trnman_public.h"
+
+/**
+   @brief drops (deletes) a table
+
+   @param  name             table's name
+
+   @return Operation status
+     @retval 0      ok
+     @retval 1      error
+*/
 
 int maria_delete_table(const char *name)
 {
@@ -25,56 +32,78 @@ int maria_delete_table(const char *name)
 #ifdef USE_RAID
   uint raid_type=0,raid_chunks=0;
 #endif
+  MARIA_HA *info;
+  myf sync_dir;
   DBUG_ENTER("maria_delete_table");
 
 #ifdef EXTRA_DEBUG
   _ma_check_table_is_closed(name,"delete");
 #endif
-  /* LOCK TODO take X-lock on table here */
+  /** @todo LOCK take X-lock on table */
+  /*
+    We need to know if this table is transactional.
+    When built with RAID support, we also need to determine if this table
+    makes use of the raid feature. If yes, we need to remove all raid
+    chunks. This is done with my_raid_delete(). Unfortunately it is
+    necessary to open the table just to check this. We use
+    'open_for_repair' to be able to open even a crashed table. If even
+    this open fails, we assume no raid configuration for this table
+    and try to remove the normal data file only. This may however
+    leave the raid chunks behind.
+  */
+  if (!(info= maria_open(name, O_RDONLY, HA_OPEN_FOR_REPAIR)))
+  {
 #ifdef USE_RAID
+    raid_type= 0;
+#endif
+    sync_dir= 0;
+  }
+  else
   {
-    MARIA_HA *info;
-    /*
-      When built with RAID support, we need to determine if this table
-      makes use of the raid feature. If yes, we need to remove all raid
-      chunks. This is done with my_raid_delete(). Unfortunately it is
-      necessary to open the table just to check this. We use
-      'open_for_repair' to be able to open even a crashed table. If even
-      this open fails, we assume no raid configuration for this table
-      and try to remove the normal data file only. This may however
-      leave the raid chunks behind.
-    */
-    if (!(info= maria_open(name, O_RDONLY, HA_OPEN_FOR_REPAIR)))
-      raid_type= 0;
-    else
-    {
-      raid_type=      info->s->base.raid_type;
-      raid_chunks=    info->s->base.raid_chunks;
-      maria_close(info);
-    }
+#ifdef USE_RAID
+    raid_type=      info->s->base.raid_type;
+    raid_chunks=    info->s->base.raid_chunks;
+#endif
+    sync_dir= (info->s->base.transactional && !info->s->temporary) ?
+      MY_SYNC_DIR : 0;
+    maria_close(info);
   }
+#ifdef USE_RAID
 #ifdef EXTRA_DEBUG
   _ma_check_table_is_closed(name,"delete");
 #endif
 #endif /* USE_RAID */
 
+  if (sync_dir)
+  {
+    /*
+      For this log record to be of any use for Recovery, we need the upper
+      MySQL layer to be crash-safe in DDLs; when it is we should reconsider
+      the moment of writing this log record, how to use it in Recovery, and
+      force the log. For now this record is only informative.
+    */
+    LSN lsn;
+    LEX_STRING log_array[TRANSLOG_INTERNAL_PARTS + 1];
+    log_array[TRANSLOG_INTERNAL_PARTS + 0].str= (char *)name;
+    log_array[TRANSLOG_INTERNAL_PARTS + 0].length= strlen(name);
+    if (unlikely(translog_write_record(&lsn, LOGREC_REDO_DROP_TABLE,
+                                       &dummy_transaction_object, NULL,
+                                       log_array[TRANSLOG_INTERNAL_PARTS +
+                                                 0].length,
+                                       sizeof(log_array)/sizeof(log_array[0]),
+                                       log_array, NULL)))
+      DBUG_RETURN(1);
+  }
+
   fn_format(from,name,"",MARIA_NAME_IEXT,MY_UNPACK_FILENAME|MY_APPEND_EXT);
-  /*
-    RECOVERY TODO log the two deletes below.
-    Then do the file deletions.
-    For this log record to be of any use for Recovery, we need the upper MySQL
-    layer to be crash-safe in DDLs; when it is we should reconsider the moment
-    of writing this log record, how to use it in Recovery, and force the log.
-    For now this record is only informative.
-  */
-  if (my_delete_with_symlink(from, MYF(MY_WME | MY_SYNC_DIR)))
+  if (my_delete_with_symlink(from, MYF(MY_WME | sync_dir)))
     DBUG_RETURN(my_errno);
   fn_format(from,name,"",MARIA_NAME_DEXT,MY_UNPACK_FILENAME|MY_APPEND_EXT);
 #ifdef USE_RAID
   if (raid_type)
-    DBUG_RETURN(my_raid_delete(from, raid_chunks, MYF(MY_WME | MY_SYNC_DIR)) ?
+    DBUG_RETURN(my_raid_delete(from, raid_chunks, MYF(MY_WME | sync_dir)) ?
                 my_errno : 0);
 #endif
-  DBUG_RETURN(my_delete_with_symlink(from, MYF(MY_WME | MY_SYNC_DIR)) ?
+  DBUG_RETURN(my_delete_with_symlink(from, MYF(MY_WME | sync_dir)) ?
               my_errno : 0);
 }

storage/maria/ma_extra.c

@@ -21,21 +21,20 @@
 static void maria_extra_keyflag(MARIA_HA *info,
                                 enum ha_extra_function function);
 
+/**
+   @brief Set options and buffers to optimize table handling
 
-/*
-  Set options and buffers to optimize table handling
+   @param  name             table's name
+   @param  info             open table
+   @param  function         operation
+   @param  extra_arg        Pointer to extra argument (normally pointer to
+                            ulong); used when function is one of:
+                            HA_EXTRA_WRITE_CACHE
+                            HA_EXTRA_CACHE
 
-  SYNOPSIS
-    maria_extra()
-    info	open table
-    function	operation
-    extra_arg	Pointer to extra argument (normally pointer to ulong)
-    		Used when function is one of:
-		HA_EXTRA_WRITE_CACHE
-		HA_EXTRA_CACHE
-  RETURN VALUES
-    0  ok
-    #  error
+   @return Operation status
+     @retval 0      ok
+     @retval !=0    error
 */
 
 int maria_extra(MARIA_HA *info, enum ha_extra_function function,
@@ -265,14 +264,24 @@ int maria_extra(MARIA_HA *info, enum ha_extra_function function,
     pthread_mutex_unlock(&THR_LOCK_maria);
     break;
   case HA_EXTRA_PREPARE_FOR_DELETE:
+    /*  QQ: suggest to rename it to "PREPARE_FOR_DROP" */
     pthread_mutex_lock(&THR_LOCK_maria);
     share->last_version= 0L;			/* Impossible version */
 #ifdef __WIN__
     /* Close the isam and data files as Win32 can't drop an open table */
     pthread_mutex_lock(&share->intern_lock);
+    /*
+      If this is Windows we remove blocks from pagecache. If not Windows we
+      don't do it, so these pages stay in the pagecache? So they may later be
+      flushed to a wrong file?
+      Or is it that this flush_pagecache_blocks() never finds any blocks? Then
+      why do we do it on Windows?
+      Don't we wait for all instances to be closed before dropping the table?
+      Do we ever do something useful here?
+      BUG?
+    */
     if (flush_pagecache_blocks(share->pagecache, &share->kfile,
-                               (function == HA_EXTRA_FORCE_REOPEN ?
-			  FLUSH_RELEASE : FLUSH_IGNORE_CHANGED)))
+                               FLUSH_IGNORE_CHANGED))
     {
       error=my_errno;
       share->changed=1;
@@ -292,9 +301,11 @@ int maria_extra(MARIA_HA *info, enum ha_extra_function function,
       info->lock_type = F_UNLCK;
     }
     if (share->kfile.file >= 0)
+    {
       _ma_decrement_open_count(info);
-    if (share->kfile.file >= 0 && my_close(share->kfile,MYF(0)))
-      error=my_errno;
+      if (my_close(share->kfile,MYF(0)))
+        error=my_errno;
+    }
     {
       LIST *list_element ;
       for (list_element=maria_open_list ;
@@ -304,6 +315,9 @@ int maria_extra(MARIA_HA *info, enum ha_extra_function function,
 	MARIA_HA *tmpinfo=(MARIA_HA*) list_element->data;
 	if (tmpinfo->s == info->s)
 	{
+          /**
+             @todo RECOVERY BUG: flush of bitmap and sync of dfile are missing
+          */
 	  if (tmpinfo->dfile.file >= 0 &&
               my_close(tmpinfo->dfile.file, MYF(0)))
 	    error = my_errno;

storage/maria/ma_init.c

@@ -53,7 +53,7 @@ void maria_end(void)
 {
   if (maria_inited)
   {
-    maria_inited= FALSE;
+    maria_inited= maria_multi_threaded= FALSE;
     ft_free_stopwords();
     trnman_destroy();
     translog_destroy();

storage/maria/ma_loghandler.h

@@ -86,13 +86,12 @@ enum translog_record_type
   LOGREC_PREPARE_WITH_UNDO_PURGE,
   LOGREC_COMMIT,
   LOGREC_COMMIT_WITH_UNDO_PURGE,
-  LOGREC_CHECKPOINT_PAGE,
-  LOGREC_CHECKPOINT_TRAN,
-  LOGREC_CHECKPOINT_TABL,
+  LOGREC_CHECKPOINT,
   LOGREC_REDO_CREATE_TABLE,
   LOGREC_REDO_RENAME_TABLE,
   LOGREC_REDO_DROP_TABLE,
-  LOGREC_REDO_TRUNCATE_TABLE,
+  LOGREC_REDO_DELETE_ALL,
+  LOGREC_REDO_REPAIR_TABLE,
   LOGREC_FILE_ID,
   LOGREC_LONG_TRANSACTION_ID,
   LOGREC_RESERVED_FUTURE_EXTENSION= 63
@@ -181,9 +180,7 @@ struct st_translog_reader_data
 };
 
 struct st_transaction;
-#ifdef	__cplusplus
-extern "C" {
-#endif
+C_MODE_START
 
 /* Records types for unittests */
 #define LOGREC_FIXED_RECORD_0LSN_EXAMPLE 1
@@ -199,13 +196,12 @@ extern my_bool translog_init(const char *directory, uint32 log_file_max_size,
 			     uint32 server_version, uint32 server_id,
 			     PAGECACHE *pagecache, uint flags);
 
-extern my_bool translog_write_record(LSN *lsn,
-                                     enum translog_record_type type,
-                                     struct st_transaction *trn,
-                                     struct st_maria_share *share,
-                                     translog_size_t rec_len,
-                                     uint part_no,
-                                     LEX_STRING *parts_data);
+extern my_bool
+translog_write_record(LSN *lsn, enum translog_record_type type,
+                      struct st_transaction *trn,
+                      struct st_maria_share *share,
+                      translog_size_t rec_len, uint part_no,
+                      LEX_STRING *parts_data, uchar *store_share_id);
 
 extern void translog_destroy();
 
@@ -232,7 +228,10 @@ extern translog_size_t translog_read_next_record_header(TRANSLOG_SCANNER_DATA
 							*scanner,
 							TRANSLOG_HEADER_BUFFER
 							*buff);
-#ifdef	__cplusplus
-}
-#endif
-
+extern void translog_lock_assert_owner();
+extern TRANSLOG_ADDRESS translog_get_horizon();
+extern int translog_assign_id_to_share(struct st_maria_share *share,
+                                       struct st_transaction *trn);
+extern void translog_deassign_id_from_share(struct st_maria_share *share);
+extern my_bool translog_inited;
+C_MODE_END

storage/maria/ma_loghandler_lsn.h

@@ -35,7 +35,7 @@ typedef TRANSLOG_ADDRESS LSN;
 /* checks LSN */
 #define LSN_VALID(L) DBUG_ASSERT((L) >= 0 && (L) < (uint64)0xFFFFFFFFFFFFFFLL)
 
-/* size of stored LSN on a disk */
+/* size of stored LSN on a disk, don't change it! */
 #define LSN_STORE_SIZE 7
 
 /* Puts LSN into buffer (dst) */
@@ -53,4 +53,12 @@ typedef TRANSLOG_ADDRESS LSN;
 
 #define LSN_REPLACE_OFFSET(L, S) (LSN_FINE_NO_PART(L) | (S))
 
+/*
+  an 8-byte type whose most significant byte is used for "flags"; 7
+  other bytes are a LSN.
+*/
+typedef LSN LSN_WITH_FLAGS;
+#define LSN_WITH_FLAGS_TO_LSN(x)   (x & ULL(0x00FFFFFFFFFFFFFF))
+#define LSN_WITH_FLAGS_TO_FLAGS(x) (x & ULL(0xFF00000000000000))
+
 #endif

storage/maria/ma_open.c

@@ -919,12 +919,23 @@ static void setup_key_functions(register MARIA_KEYDEF *keyinfo)
 }
 
 
-/*
-   Function to save and store the header in the index file (.MYI)
+/**
+   @brief Function to save and store the header in the index file (.MYI)
+
+   @param  file            descriptor of the index file to write
+   @param  state           state information to write to the file
+   @param  pWrite          bitmap (determines the amount of information to
+                           write, and if my_write() or my_pwrite() should be
+                           used)
+
+   @return Operation status
+     @retval 0      OK
+     @retval 1      Error
 */
 
 uint _ma_state_info_write(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;
@@ -935,6 +946,11 @@ uint _ma_state_info_write(File file, MARIA_STATE_INFO *state, uint pWrite)
 
   /* open_count must be first because of _ma_mark_file_changed ! */
   mi_int2store(ptr,state->open_count);			ptr+= 2;
+  /*
+    if you change the offset of this LSN inside the file, fix
+    ma_create + ma_rename + ma_delete_all + backward-compatibility.
+  */
+  lsn_store(ptr, state->create_rename_lsn);		ptr+= LSN_STORE_SIZE;
   *ptr++= (uchar)state->changed;
   *ptr++= state->sortkey;
   mi_rowstore(ptr,state->state.records);		ptr+= 8;
@@ -959,6 +975,7 @@ uint _ma_state_info_write(File file, MARIA_STATE_INFO *state, uint pWrite)
   {
     mi_sizestore(ptr,state->key_root[i]);		ptr+= 8;
   }
+  /** @todo RECOVERY key_del is a problem for recovery */
   mi_sizestore(ptr,state->key_del);	        	ptr+= 8;
   if (pWrite & 2)				/* From maria_chk */
   {
@@ -994,6 +1011,7 @@ byte *_ma_state_info_read(byte *ptr, MARIA_STATE_INFO *state)
   key_parts= mi_uint2korr(state->header.key_parts);
 
   state->open_count = mi_uint2korr(ptr);		ptr+= 2;
+  state->create_rename_lsn= lsn_korr(ptr);		ptr+= LSN_STORE_SIZE;
   state->changed= 					(my_bool) *ptr++;
   state->sortkey= 					(uint) *ptr++;
   state->state.records= mi_rowkorr(ptr);		ptr+= 8;

storage/maria/ma_pagecache.h

@@ -239,6 +239,7 @@ extern my_bool pagecache_delete_pages(PAGECACHE *pagecache,
 extern void end_pagecache(PAGECACHE *keycache, my_bool cleanup);
 extern my_bool pagecache_collect_changed_blocks_with_lsn(PAGECACHE *pagecache,
                                                          LEX_STRING *str,
+                                                         LSN *min_lsn,
                                                          LSN *max_lsn);
 extern int reset_pagecache_counters(const char *name, PAGECACHE *pagecache);
 

storage/maria/ma_panic.c

@@ -52,7 +52,12 @@ int maria_panic(enum ha_panic_function flag)
     info=(MARIA_HA*) list_element->data;
     switch (flag) {
     case HA_PANIC_CLOSE:
-      pthread_mutex_unlock(&THR_LOCK_maria);	/* Not exactly right... */
+      /*
+        If bad luck (if some tables would be used now, which normally does not
+        happen in MySQL), as we release the mutex, the list may change and so
+        we may crash.
+      */
+      pthread_mutex_unlock(&THR_LOCK_maria);
       if (maria_close(info))
 	error=my_errno;
       pthread_mutex_lock(&THR_LOCK_maria);

storage/maria/ma_range.c

@@ -29,25 +29,22 @@ static uint _ma_keynr(MARIA_HA *info, MARIA_KEYDEF *keyinfo, byte *page,
 		      byte *keypos, uint *ret_max_key);
 
 
-/*
-  Estimate how many records there is in a given range
+/**
+   @brief Estimate how many records there is in a given range
 
-  SYNOPSIS
-    maria_records_in_range()
-    info		MARIA handler
-    inx			Index to use
-    min_key		Min key. Is = 0 if no min range
-    max_key		Max key. Is = 0 if no max range
+   @param  info            MARIA handler
+   @param  inx             Index to use
+   @param  min_key         Min key. Is = 0 if no min range
+   @param  max_key         Max key. Is = 0 if no max range
 
-  NOTES
-    We should ONLY return 0 if there is no rows in range
+   @note
+     We should ONLY return 0 if there is no rows in range
 
-  RETURN
-    HA_POS_ERROR  error (or we can't estimate number of rows)
-    number	  Estimated number of rows
+   @return Estimated number of rows or error
+     @retval HA_POS_ERROR  error (or we can't estimate number of rows)
+     @retval number        Estimated number of rows
 */
 
-
 ha_rows maria_records_in_range(MARIA_HA *info, int inx, key_range *min_key,
                             key_range *max_key)
 {
@@ -115,6 +112,13 @@ ha_rows maria_records_in_range(MARIA_HA *info, int inx, key_range *min_key,
     rw_unlock(&info->s->key_root_lock[inx]);
   fast_ma_writeinfo(info);
 
+  /**
+     @todo LOCK
+     If res==0 (no rows), if we need to guarantee repeatability of the search,
+     we will need to set a next-key lock in this statement.
+     Also SELECT COUNT(*)...
+  */
+
   DBUG_PRINT("info",("records: %ld",(ulong) (res)));
   DBUG_RETURN(res);
 }

storage/maria/ma_rename.c

@@ -18,6 +18,18 @@
 */
 
 #include "ma_fulltext.h"
+#include "trnman_public.h"
+
+/**
+   @brief renames a table
+
+   @param  old_name        current name of table
+   @param  new_name        table should be renamed to this name
+
+   @return Operation status
+     @retval 0      OK
+     @retval !=0    Error
+*/
 
 int maria_rename(const char *old_name, const char *new_name)
 {
@@ -26,22 +38,73 @@ int maria_rename(const char *old_name, const char *new_name)
 #ifdef USE_RAID
   uint raid_type=0,raid_chunks=0;
 #endif
+  MARIA_HA *info;
+  MARIA_SHARE *share;
+  myf sync_dir;
   DBUG_ENTER("maria_rename");
 
 #ifdef EXTRA_DEBUG
   _ma_check_table_is_closed(old_name,"rename old_table");
   _ma_check_table_is_closed(new_name,"rename new table2");
 #endif
-  /* LOCK TODO take X-lock on table here */
+  /** @todo LOCK take X-lock on table */
+  if (!(info= maria_open(old_name, O_RDWR, HA_OPEN_FOR_REPAIR)))
+    DBUG_RETURN(my_errno);
+  share= info->s;
 #ifdef USE_RAID
+  raid_type =      share->base.raid_type;
+  raid_chunks =    share->base.raid_chunks;
+#endif
+
+  sync_dir= (share->base.transactional && !share->temporary) ?
+    MY_SYNC_DIR : 0;
+  if (sync_dir)
   {
-    MARIA_HA *info;
-    if (!(info=maria_open(old_name, O_RDONLY, 0)))
-      DBUG_RETURN(my_errno);
-    raid_type =      info->s->base.raid_type;
-    raid_chunks =    info->s->base.raid_chunks;
-    maria_close(info);
+    uchar log_data[LSN_STORE_SIZE];
+    LEX_STRING log_array[TRANSLOG_INTERNAL_PARTS + 3];
+    uint old_name_len= strlen(old_name), new_name_len= strlen(new_name);
+    int2store(log_data, old_name_len);
+    int2store(log_data + 2, new_name_len);
+    log_array[TRANSLOG_INTERNAL_PARTS + 0].str= log_data;
+    log_array[TRANSLOG_INTERNAL_PARTS + 0].length= 2 + 2;
+    log_array[TRANSLOG_INTERNAL_PARTS + 1].str= (char *)old_name;
+    log_array[TRANSLOG_INTERNAL_PARTS + 1].length= old_name_len;
+    log_array[TRANSLOG_INTERNAL_PARTS + 2].str= (char *)new_name;
+    log_array[TRANSLOG_INTERNAL_PARTS + 2].length= new_name_len;
+    /*
+      For this record to be of any use for Recovery, we need the upper
+      MySQL layer to be crash-safe, which it is not now (that would require
+      work using the ddl_log of sql/sql_table.cc); when it is, we should
+      reconsider the moment of writing this log record (before or after op,
+      under THR_LOCK_maria or not...), how to use it in Recovery, and force
+      the log. For now this record is just informative.
+    */
+    if (unlikely(translog_write_record(&share->state.create_rename_lsn,
+                                       LOGREC_REDO_RENAME_TABLE,
+                                       &dummy_transaction_object, NULL,
+                                       2 + 2 + old_name_len + new_name_len,
+                                       sizeof(log_array)/sizeof(log_array[0]),
+                                       log_array, NULL)))
+    {
+      maria_close(info);
+      DBUG_RETURN(1);
+    }
+    /*
+      store LSN into file, needed for Recovery to not be confused if a
+      RENAME happened (applying REDOs to the wrong table).
+    */
+    lsn_store(log_data, share->state.create_rename_lsn);
+    if (my_pwrite(share->kfile.file, log_data, sizeof(log_data),
+                  sizeof(share->state.header) + 2, MYF(MY_NABP)) ||
+        my_sync(share->kfile.file, MYF(MY_WME)))
+    {
+      maria_close(info);
+      DBUG_RETURN(1);
+    }
   }
+
+  maria_close(info);
+#ifdef USE_RAID
 #ifdef EXTRA_DEBUG
   _ma_check_table_is_closed(old_name,"rename raidcheck");
 #endif
@@ -49,29 +112,18 @@ int maria_rename(const char *old_name, const char *new_name)
 
   fn_format(from,old_name,"",MARIA_NAME_IEXT,MY_UNPACK_FILENAME|MY_APPEND_EXT);
   fn_format(to,new_name,"",MARIA_NAME_IEXT,MY_UNPACK_FILENAME|MY_APPEND_EXT);
-  /*
-    RECOVERY TODO log the two renames below. Update
-    ZeroDirtyPagesLSN of the table on disk (=> sync the files), this is
-    needed so that Recovery does not pick a wrong table.
-    Then do the file renames.
-    For this log record to be of any use for Recovery, we need the upper MySQL
-    layer to be crash-safe in DDLs; when it is we should reconsider the moment
-    of writing this log record, how to use it in Recovery, and force the log.
-    For now this record is only informative. But ZeroDirtyPagesLSN is
-    critically needed!
-  */
-  if (my_rename_with_symlink(from, to, MYF(MY_WME | MY_SYNC_DIR)))
+  if (my_rename_with_symlink(from, to, MYF(MY_WME | sync_dir)))
     DBUG_RETURN(my_errno);
   fn_format(from,old_name,"",MARIA_NAME_DEXT,MY_UNPACK_FILENAME|MY_APPEND_EXT);
   fn_format(to,new_name,"",MARIA_NAME_DEXT,MY_UNPACK_FILENAME|MY_APPEND_EXT);
 #ifdef USE_RAID
   if (raid_type)
     data_file_rename_error= my_raid_rename(from, to, raid_chunks,
-                                           MYF(MY_WME | MY_SYNC_DIR));
+                                           MYF(MY_WME | sync_dir));
   else
 #endif
     data_file_rename_error=
-      my_rename_with_symlink(from, to, MYF(MY_WME | MY_SYNC_DIR));
+      my_rename_with_symlink(from, to, MYF(MY_WME | sync_dir));
   if (data_file_rename_error)
   {
     /*
@@ -81,7 +133,7 @@ int maria_rename(const char *old_name, const char *new_name)
     data_file_rename_error= my_errno;
     fn_format(from, old_name, "", MARIA_NAME_IEXT, MYF(MY_UNPACK_FILENAME|MY_APPEND_EXT));
     fn_format(to, new_name, "", MARIA_NAME_IEXT, MYF(MY_UNPACK_FILENAME|MY_APPEND_EXT));
-    my_rename_with_symlink(to, from, MYF(MY_WME | MY_SYNC_DIR));
+    my_rename_with_symlink(to, from, MYF(MY_WME | sync_dir));
   }
   DBUG_RETURN(data_file_rename_error);
 

storage/maria/ma_static.c

@@ -47,7 +47,13 @@ PAGECACHE *maria_pagecache= &maria_pagecache_var;
 PAGECACHE maria_log_pagecache_var;
 PAGECACHE *maria_log_pagecache= &maria_log_pagecache_var;
 
-/* For using maria externally */
+/**
+   @brief when transactionality does not matter we can use this transaction
+
+   Used in external programs like ma_test*, and also internally inside
+   libmaria when there is no transaction around and the operation isn't
+   transactional (CREATE/DROP/RENAME/OPTIMIZE/REPAIR).
+*/
 TRN dummy_transaction_object;
 
 /* Enough for comparing if number is zero */

storage/maria/maria_def.h

@@ -93,6 +93,7 @@ typedef struct st_maria_state_info
   uint sortkey;				/* sorted by this key (not used) */
   uint open_count;
   uint8 changed;			/* Changed since mariachk */
+  LSN create_rename_lsn;    /**< LSN when table was last created/renamed */
 
   /* the following isn't saved on disk */
   uint state_diff_length;		/* Should be 0 */
@@ -101,7 +102,8 @@ typedef struct st_maria_state_info
 } MARIA_STATE_INFO;
 
 
-#define MARIA_STATE_INFO_SIZE	(24 + 4 + 11*8 + 4*4 + 8 + 3*4 + 5*8)
+#define MARIA_STATE_INFO_SIZE	\
+  (24 + LSN_STORE_SIZE + 4 + 11*8 + 4*4 + 8 + 3*4 + 5*8)
 #define MARIA_STATE_KEY_SIZE	8
 #define MARIA_STATE_KEYBLOCK_SIZE  8
 #define MARIA_STATE_KEYSEG_SIZE	4
@@ -229,6 +231,7 @@ typedef struct st_maria_share
   PAGECACHE *pagecache;			/* ref to the current key cache */
   MARIA_DECODE_TREE *decode_trees;
   uint16 *decode_tables;
+  uint16 id; /**< 2-byte id by which log records refer to the table */
   /* Called the first time the table instance is opened */
   my_bool (*once_init)(struct st_maria_share *, File);
   /* Called when the last instance of the table is closed */
@@ -889,6 +892,7 @@ volatile int *_ma_killed_ptr(HA_CHECK *param);
 void _ma_check_print_error _VARARGS((HA_CHECK *param, const char *fmt, ...));
 void _ma_check_print_warning _VARARGS((HA_CHECK *param, const char *fmt, ...));
 void _ma_check_print_info _VARARGS((HA_CHECK *param, const char *fmt, ...));
+int  _ma_repair_write_log_record(const HA_CHECK *param, MARIA_HA *info);
 C_MODE_END
 
 int _ma_flush_pending_blocks(MARIA_SORT_PARAM *param);

storage/maria/trnman.c

@@ -52,6 +52,7 @@ static my_atomic_rwlock_t LOCK_short_trid_to_trn, LOCK_pool;
 
 /*
   Simple interface functions
+  QQ: if they stay so simple, should we make them inline?
 */
 
 uint trnman_increment_locked_tables(TRN *trn)
@@ -343,6 +344,9 @@ int trnman_end_trn(TRN *trn, my_bool commit)
   LF_PINS *pins= trn->pins;
   DBUG_ENTER("trnman_end_trn");
 
+  DBUG_ASSERT(trn->rec_lsn == 0);
+  /* if a rollback, all UNDO records should have been executed */
+  DBUG_ASSERT(commit || trn->undo_lsn == 0);
   DBUG_PRINT("info", ("pthread_mutex_lock LOCK_trn_list"));
   pthread_mutex_lock(&LOCK_trn_list);
 
@@ -379,8 +383,6 @@ int trnman_end_trn(TRN *trn, my_bool commit)
   /*
     if transaction is committed and it was not the only active transaction -
     add it to the committed list (which is used for read-from relation)
-    TODO check in the condition below that a transaction have made some
-    changes, was not read-only. Something like '&& UndoLSN != 0'
   */
   if (commit && active_list_min.next != &active_list_max)
   {
@@ -390,6 +392,19 @@ int trnman_end_trn(TRN *trn, my_bool commit)
     trnman_committed_transactions++;
 
     res= lf_hash_insert(&trid_to_committed_trn, pins, &trn);
+    /*
+      By going on with life is res<0, we let other threads block on
+      our rows (because they will never see us committed in
+      trid_to_committed_trn) until they timeout. Though correct, this is not a
+      good situation:
+      - if connection reconnects and wants to check if its rows have been
+      committed, it will not be able to do that (it will just lock on them) so
+      connection stays permanently in doubt
+      - internal structures trid_to_committed_trn and committed_list are
+      desynchronized.
+      So we should take Maria down immediately, the two problems being
+      automatically solved at restart.
+    */
     DBUG_ASSERT(res <= 0);
   }
   if (res)
@@ -526,71 +541,133 @@ void trnman_rollback_statement(TRN *trn __attribute__ ((unused)))
 }
 
 
-/*
-  Allocates two buffers and stores in them some information about transactions
-  of the active list (into the first buffer) and of the committed list (into
-  the second buffer).
-
-  SYNOPSIS
-    trnman_collect_transactions()
-    str_act    (OUT) pointer to a LEX_STRING where the allocated buffer, and
-               its size, will be put
-    str_com    (OUT) pointer to a LEX_STRING where the allocated buffer, and
-               its size, will be put
+/**
+   @brief Allocates buffers and stores in them some info about transactions
 
+   Does the allocation because the caller cannot know the size itself.
+   Memory freeing is to be done by the caller (if the "str" member of the
+   LEX_STRING is not NULL).
+   The caller has the intention of doing checkpoints.
 
-  DESCRIPTION
-    Does the allocation because the caller cannot know the size itself.
-    Memory freeing is to be done by the caller (if the "str" member of the
-    LEX_STRING is not NULL).
-    The caller has the intention of doing checkpoints.
+   @param[out]  str_act    pointer to where the allocated buffer,
+                           and its size, will be put; buffer will be filled
+                           with info about active transactions
+   @param[out]  str_com    pointer to where the allocated buffer,
+                           and its size, will be put; buffer will be filled
+                           with info about committed transactions
+   @param[out]  min_first_undo_lsn pointer to where the minimum
+                           first_undo_lsn of all transactions will be put
 
-  RETURN
-    0 on success
-    1 on error
+   @return Operation status
+     @retval 0      OK
+     @retval 1      Error
 */
-my_bool trnman_collect_transactions(LEX_STRING *str_act, LEX_STRING *str_com)
+
+my_bool trnman_collect_transactions(LEX_STRING *str_act, LEX_STRING *str_com,
+                                    LSN *min_rec_lsn, LSN *min_first_undo_lsn)
 {
   my_bool error;
   TRN *trn;
   char *ptr;
+  uint stored_transactions= 0;
+  LSN minimum_rec_lsn= ULONGLONG_MAX, minimum_first_undo_lsn= ULONGLONG_MAX;
   DBUG_ENTER("trnman_collect_transactions");
 
   DBUG_ASSERT((NULL == str_act->str) && (NULL == str_com->str));
 
+  /* validate the use of read_non_atomic() in general: */
+  compile_time_assert((sizeof(LSN) == 8) && (sizeof(LSN_WITH_FLAGS) == 8));
+
   DBUG_PRINT("info", ("pthread_mutex_lock LOCK_trn_list"));
   pthread_mutex_lock(&LOCK_trn_list);
-  str_act->length= 8+(6+2+7+7+7)*trnman_active_transactions;
-  str_com->length= 8+(6+7+7)*trnman_committed_transactions;
+  str_act->length= 2 + /* number of active transactions */
+    LSN_STORE_SIZE + /* minimum of their rec_lsn */
+    (6 + /* long id */
+     2 + /* short id */
+     LSN_STORE_SIZE + /* undo_lsn */
+#ifdef MARIA_VERSIONING /* not enabled yet */
+     LSN_STORE_SIZE + /* undo_purge_lsn */
+#endif
+     LSN_STORE_SIZE /* first_undo_lsn */
+     ) * trnman_active_transactions;
+  str_com->length= 8 + /* number of committed transactions */
+    (6 + /* long id */
+#ifdef MARIA_VERSIONING /* not enabled yet */
+     LSN_STORE_SIZE + /* undo_purge_lsn */
+#endif
+     LSN_STORE_SIZE /* first_undo_lsn */
+     ) * trnman_committed_transactions;
   if ((NULL == (str_act->str= my_malloc(str_act->length, MYF(MY_WME)))) ||
       (NULL == (str_com->str= my_malloc(str_com->length, MYF(MY_WME)))))
     goto err;
   /* First, the active transactions */
-  ptr= str_act->str;
-  int8store(ptr, (ulonglong)trnman_active_transactions);
-  ptr+= 8;
+  ptr= str_act->str + 2 + LSN_STORE_SIZE;
   for (trn= active_list_min.next; trn != &active_list_max; trn= trn->next)
   {
     /*
-      trns with a short trid of 0 are not initialized; Recovery will recognize
-      this and ignore them.
-      State is not needed for now (only when we supported prepared trns).
-      For LSNs, Sanja will soon push lsn7store.
+      trns with a short trid of 0 are not even initialized, we can ignore
+      them. trns with undo_lsn==0 have done no writes, we can ignore them
+      too. XID not needed now.
     */
+    uint sid;
+    LSN rec_lsn, undo_lsn, first_undo_lsn;
+    if ((sid= trn->short_id) == 0)
+    {
+      /*
+        Not even inited, has done nothing. Or it is the
+        dummy_transaction_object, which does only non-transactional
+        immediate-sync operations (CREATE/DROP/RENAME/REPAIR TABLE), and so
+        can be forgotten for Checkpoint.
+      */
+      continue;
+    }
+#ifndef MARIA_CHECKPOINT
+/*
+  in the checkpoint patch (not yet ready) we will have a real implementation
+  of lsn_read_non_atomic(); for now it's not needed
+*/
+#define lsn_read_non_atomic(A) (A)
+#endif
+      /* needed for low-water mark calculation */
+    if (((rec_lsn= lsn_read_non_atomic(trn->rec_lsn)) > 0) &&
+        (cmp_translog_addr(rec_lsn, minimum_rec_lsn) < 0))
+      minimum_rec_lsn= rec_lsn;
+    /*
+      trn may have logged REDOs but not yet UNDO, that's why we read rec_lsn
+      before deciding to ignore if undo_lsn==0.
+    */
+    if  ((undo_lsn= trn->undo_lsn) == 0) /* trn can be forgotten */
+      continue;
+    stored_transactions++;
     int6store(ptr, trn->trid);
     ptr+= 6;
-    int2store(ptr, trn->short_id);
+    int2store(ptr, sid);
     ptr+= 2;
-    /* needed for rollback */
-    /* lsn7store(ptr, trn->undo_lsn); */
-    ptr+= 7;
-    /* needed for purge */
-    /* lsn7store(ptr, trn->undo_purge_lsn); */
-    ptr+= 7;
+    lsn_store(ptr, undo_lsn); /* needed for rollback */
+    ptr+= LSN_STORE_SIZE;
+#ifdef MARIA_VERSIONING /* not enabled yet */
+    /* to know where purging should start (last delete of this trn) */
+    lsn_store(ptr, trn->undo_purge_lsn);
+    ptr+= LSN_STORE_SIZE;
+#endif
     /* needed for low-water mark calculation */
-    /* lsn7store(ptr, read_non_atomic(&trn->first_undo_lsn)); */
-    ptr+= 7;
+    if (((first_undo_lsn= lsn_read_non_atomic(trn->first_undo_lsn)) > 0) &&
+        (cmp_translog_addr(first_undo_lsn, minimum_first_undo_lsn) < 0))
+      minimum_first_undo_lsn= first_undo_lsn;
+    lsn_store(ptr, first_undo_lsn);
+    ptr+= LSN_STORE_SIZE;
+    /**
+       @todo RECOVERY: add a comment explaining why we can dirtily read some
+       vars, inspired by the text of "assumption 8" in WL#3072
+    */
   }
+  str_act->length= ptr - str_act->str; /* as we maybe over-estimated */
+  ptr= str_act->str;
+  int2store(ptr, stored_transactions);
+  ptr+= 2;
+  /* this LSN influences how REDOs for any page can be ignored by Recovery */
+  lsn_store(ptr, minimum_rec_lsn);
+  /* one day there will also be a list of prepared transactions */
   /* do the same for committed ones */
   ptr= str_com->str;
   int8store(ptr, (ulonglong)trnman_committed_transactions);
@@ -598,18 +675,26 @@ my_bool trnman_collect_transactions(LEX_STRING *str_act, LEX_STRING *str_com)
   for (trn= committed_list_min.next; trn != &committed_list_max;
        trn= trn->next)
   {
+    LSN first_undo_lsn;
     int6store(ptr, trn->trid);
     ptr+= 6;
-    /* mi_int7store(ptr, trn->undo_purge_lsn); */
-    ptr+= 7;
-    /* mi_int7store(ptr, read_non_atomic(&trn->first_undo_lsn)); */
-    ptr+= 7;
+#ifdef MARIA_VERSIONING /* not enabled yet */
+    lsn_store(ptr, trn->undo_purge_lsn);
+    ptr+= LSN_STORE_SIZE;
+#endif
+    first_undo_lsn= LSN_WITH_FLAGS_TO_LSN(trn->first_undo_lsn);
+    if (cmp_translog_addr(first_undo_lsn, minimum_first_undo_lsn) < 0)
+      minimum_first_undo_lsn= first_undo_lsn;
+    lsn_store(ptr, first_undo_lsn);
+    ptr+= LSN_STORE_SIZE;
   }
   /*
     TODO: if we see there exists no transaction (active and committed) we can
     tell the lock-free structures to do some freeing (my_free()).
   */
   error= 0;
+  *min_rec_lsn= minimum_rec_lsn;
+  *min_first_undo_lsn= minimum_first_undo_lsn;
   goto end;
 err:
   error= 1;

storage/maria/trnman.h

@@ -45,12 +45,13 @@ struct st_transaction
   LF_PINS             *pins;
   TrID                 trid, min_read_from, commit_trid;
   TRN                 *next, *prev;
-  LSN		       rec_lsn, undo_lsn, first_undo_lsn;
+  LSN		       rec_lsn, undo_lsn;
+  LSN_WITH_FLAGS       first_undo_lsn;
   uint                 locked_tables;
   /* Note! if locks.loid is 0, trn is NOT initialized */
 };
 
-TRN dummy_transaction_object;
+#define TRANSACTION_LOGGED_LONG_ID ULL(0x8000000000000000)
 
 C_MODE_END
 

storage/maria/trnman_public.h

@@ -20,6 +20,8 @@
   to include my_atomic.h in C++ code.
 */
 
+#include "ma_loghandler_lsn.h"
+
 C_MODE_START
 typedef uint64 TrID; /* our TrID is 6 bytes */
 typedef struct st_transaction TRN;
@@ -27,6 +29,7 @@ typedef struct st_transaction TRN;
 #define SHORT_TRID_MAX 65535
 
 extern uint trnman_active_transactions, trnman_allocated_transactions;
+extern TRN dummy_transaction_object;
 
 int trnman_init(void);
 void trnman_destroy(void);
@@ -39,7 +42,9 @@ void trnman_free_trn(TRN *trn);
 int trnman_can_read_from(TRN *trn, TrID trid);
 void trnman_new_statement(TRN *trn);
 void trnman_rollback_statement(TRN *trn);
-my_bool trnman_collect_transactions(LEX_STRING *str_act, LEX_STRING *str_com);
+my_bool trnman_collect_transactions(LEX_STRING *str_act, LEX_STRING *str_com,
+                                    LSN *min_rec_lsn,
+                                    LSN *min_first_undo_lsn);
 
 uint trnman_increment_locked_tables(TRN *trn);
 uint trnman_decrement_locked_tables(TRN *trn);

storage/maria/unittest/ma_test_loghandler-t.c

@@ -196,7 +196,7 @@ int main(int argc __attribute__((unused)), char *argv[])
   if (translog_write_record(&lsn,
                             LOGREC_FIXED_RECORD_0LSN_EXAMPLE,
                             trn, NULL,
-                            6, TRANSLOG_INTERNAL_PARTS + 1, parts))
+                            6, TRANSLOG_INTERNAL_PARTS + 1, parts, NULL))
   {
     fprintf(stderr, "Can't write record #%lu\n", (ulong) 0);
     translog_destroy();
@@ -218,7 +218,7 @@ int main(int argc __attribute__((unused)), char *argv[])
       parts[TRANSLOG_INTERNAL_PARTS + 1].str= NULL;
       parts[TRANSLOG_INTERNAL_PARTS + 1].length= 0;
       if (translog_write_record(&lsn, LOGREC_FIXED_RECORD_1LSN_EXAMPLE,
-                                trn, NULL, LSN_STORE_SIZE, 0, parts))
+                                trn, NULL, LSN_STORE_SIZE, 0, parts, NULL))
       {
         fprintf(stderr, "1 Can't write reference defore record #%lu\n",
                 (ulong) i);
@@ -238,7 +238,7 @@ int main(int argc __attribute__((unused)), char *argv[])
       if (translog_write_record(&lsn,
                                 LOGREC_VARIABLE_RECORD_1LSN_EXAMPLE,
                                 trn, NULL, 0, TRANSLOG_INTERNAL_PARTS + 2,
-                                parts))
+                                parts, NULL))
       {
         fprintf(stderr, "1 Can't write var reference defore record #%lu\n",
                 (ulong) i);
@@ -257,7 +257,7 @@ int main(int argc __attribute__((unused)), char *argv[])
       if (translog_write_record(&lsn,
                                 LOGREC_FIXED_RECORD_2LSN_EXAMPLE,
                                 trn, NULL,
-                                23, TRANSLOG_INTERNAL_PARTS + 1, parts))
+                                23, TRANSLOG_INTERNAL_PARTS + 1, parts, NULL))
       {
         fprintf(stderr, "0 Can't write reference defore record #%lu\n",
                 (ulong) i);
@@ -277,7 +277,7 @@ int main(int argc __attribute__((unused)), char *argv[])
       if (translog_write_record(&lsn,
                                 LOGREC_VARIABLE_RECORD_2LSN_EXAMPLE,
                                 trn, NULL, 14 + rec_len,
-                                TRANSLOG_INTERNAL_PARTS + 2, parts))
+                                TRANSLOG_INTERNAL_PARTS + 2, parts, NULL))
       {
         fprintf(stderr, "0 Can't write var reference defore record #%lu\n",
                 (ulong) i);
@@ -294,7 +294,7 @@ int main(int argc __attribute__((unused)), char *argv[])
                               LOGREC_FIXED_RECORD_0LSN_EXAMPLE,
                               trn, NULL, 6,
                               TRANSLOG_INTERNAL_PARTS + 1,
-                              parts))
+                              parts, NULL))
     {
       fprintf(stderr, "Can't write record #%lu\n", (ulong) i);
       translog_destroy();
@@ -313,7 +313,7 @@ int main(int argc __attribute__((unused)), char *argv[])
                               LOGREC_VARIABLE_RECORD_0LSN_EXAMPLE,
                               trn, NULL, rec_len,
                               TRANSLOG_INTERNAL_PARTS + 1,
-                              parts))
+                              parts, NULL))
     {
       fprintf(stderr, "Can't write variable record #%lu\n", (ulong) i);
       translog_destroy();

storage/maria/unittest/ma_test_loghandler_multigroup-t.c

@@ -192,7 +192,7 @@ int main(int argc __attribute__((unused)), char *argv[])
   trn->short_id= 0;
   if (translog_write_record(&lsn, LOGREC_FIXED_RECORD_0LSN_EXAMPLE,
                             trn, NULL,
-                            6, TRANSLOG_INTERNAL_PARTS + 1, parts))
+                            6, TRANSLOG_INTERNAL_PARTS + 1, parts, NULL))
   {
     fprintf(stderr, "Can't write record #%lu\n", (ulong) 0);
     translog_destroy();
@@ -214,7 +214,7 @@ int main(int argc __attribute__((unused)), char *argv[])
                                 LOGREC_FIXED_RECORD_1LSN_EXAMPLE,
                                 trn, NULL,
                                 LSN_STORE_SIZE,
-                                TRANSLOG_INTERNAL_PARTS + 1, parts))
+                                TRANSLOG_INTERNAL_PARTS + 1, parts, NULL))
       {
         fprintf(stderr, "1 Can't write reference before record #%lu\n",
                 (ulong) i);
@@ -234,7 +234,7 @@ int main(int argc __attribute__((unused)), char *argv[])
                                 LOGREC_VARIABLE_RECORD_1LSN_EXAMPLE,
                                 trn, NULL, LSN_STORE_SIZE + rec_len,
                                 TRANSLOG_INTERNAL_PARTS + 2,
-                                parts))
+                                parts, NULL))
       {
         fprintf(stderr, "1 Can't write var reference before record #%lu\n",
                 (ulong) i);
@@ -255,7 +255,7 @@ int main(int argc __attribute__((unused)), char *argv[])
                                 LOGREC_FIXED_RECORD_2LSN_EXAMPLE,
                                 trn, NULL, 23,
                                 TRANSLOG_INTERNAL_PARTS + 1,
-                                parts))
+                                parts, NULL))
       {
         fprintf(stderr, "0 Can't write reference before record #%lu\n",
                 (ulong) i);
@@ -276,7 +276,7 @@ int main(int argc __attribute__((unused)), char *argv[])
                                 LOGREC_VARIABLE_RECORD_2LSN_EXAMPLE,
                                 trn, NULL, LSN_STORE_SIZE * 2 + rec_len,
                                 TRANSLOG_INTERNAL_PARTS + 2,
-                                parts))
+                                parts, NULL))
       {
         fprintf(stderr, "0 Can't write var reference before record #%lu\n",
                 (ulong) i);
@@ -293,7 +293,7 @@ int main(int argc __attribute__((unused)), char *argv[])
     if (translog_write_record(&lsn,
                               LOGREC_FIXED_RECORD_0LSN_EXAMPLE,
                               trn, NULL, 6,
-                              TRANSLOG_INTERNAL_PARTS + 1, parts))
+                              TRANSLOG_INTERNAL_PARTS + 1, parts, NULL))
     {
       fprintf(stderr, "Can't write record #%lu\n", (ulong) i);
       translog_destroy();
@@ -311,7 +311,7 @@ int main(int argc __attribute__((unused)), char *argv[])
     if (translog_write_record(&lsn,
                               LOGREC_VARIABLE_RECORD_0LSN_EXAMPLE,
                               trn, NULL, rec_len,
-                              TRANSLOG_INTERNAL_PARTS + 1, parts))
+                              TRANSLOG_INTERNAL_PARTS + 1, parts, NULL))
     {
       fprintf(stderr, "Can't write variable record #%lu\n", (ulong) i);
       translog_destroy();

storage/maria/unittest/ma_test_loghandler_multithread-t.c

@@ -137,7 +137,7 @@ void writer(int num)
     if (translog_write_record(&lsn,
                               LOGREC_FIXED_RECORD_0LSN_EXAMPLE,
                               &trn, NULL, 6, TRANSLOG_INTERNAL_PARTS + 1,
-                              parts))
+                              parts, NULL))
     {
       fprintf(stderr, "Can't write LOGREC_FIXED_RECORD_0LSN_EXAMPLE record #%lu "
               "thread %i\n", (ulong) i, num);
@@ -154,7 +154,7 @@ void writer(int num)
                               LOGREC_VARIABLE_RECORD_0LSN_EXAMPLE,
                               &trn, NULL,
                               len, TRANSLOG_INTERNAL_PARTS + 1,
-                              parts))
+                              parts, NULL))
     {
       fprintf(stderr, "Can't write variable record #%lu\n", (ulong) i);
       translog_destroy();
@@ -303,7 +303,7 @@ int main(int argc __attribute__((unused)),
                               LOGREC_FIXED_RECORD_0LSN_EXAMPLE,
                               &dummy_transaction_object, NULL, 6,
                               TRANSLOG_INTERNAL_PARTS + 1,
-                              parts))
+                              parts, NULL))
     {
       fprintf(stderr, "Can't write the first record\n");
       translog_destroy();

storage/maria/unittest/ma_test_loghandler_pagecache-t.c

@@ -94,7 +94,7 @@ int main(int argc __attribute__((unused)), char *argv[])
                             LOGREC_FIXED_RECORD_0LSN_EXAMPLE,
                             &dummy_transaction_object, NULL, 6,
                             TRANSLOG_INTERNAL_PARTS + 1,
-                            parts))
+                            parts, NULL))
   {
     fprintf(stderr, "Can't write record #%lu\n", (ulong) 0);
     translog_destroy();