This patch is a collection of patches from from Sanja, Sergei and Monty.

Added logging and pinning of pages to block format.
Integration of transaction manager, log handler.
Better page cache intergration
Split trnman.h into two files, so that we don't have to include my_atomic.h into C++ programs.
Renaming of structures, more comments, more debugging etc.
Fixed problem with small head block + long varchar.
Added extra argument to delete_record() and update_record() (needed for UNDO logging)
Small changes to interface of pagecache and log handler.
Change initialization of log_record_type_descriptors to not be depending on enum order.
Use array of LEX_STRING's to send data to log handler
Added 'dummy' transaction option to MARIA_INFO so that we can always assume 'trn' exists.


include/lf.h:
  Interface fixes
  Rename of structures
  (Patch from Sergei via Sanja)
include/my_atomic.h:
  More comments
include/my_global.h:
  Added MY_ERRPTR
include/pagecache.h:
  Added undo LSN when unlocking pages
mysql-test/r/maria.result:
  Updated results
mysql-test/t/maria.test:
  Added autocommit around lock tables
  (Patch from Sanja)
mysys/lf_alloc-pin.c:
  Post-review fixes, simple optimizations
  More comments
  Struct slot renames
  Check amount of memory on stack
  (Patch from Sergei)
mysys/lf_dynarray.c:
  More comments
mysys/lf_hash.c:
  More comments
  After review fixes
  (Patch from Sergei)
storage/maria/ha_maria.cc:
  Split trnman.h into two files, so that we don't have to include my_atomic.h into the .cc program.
  (Temporary fix to avoid bug in gcc)
  Move out all deferencing of the transaction structure.
  Transaction manager integrated (Patch from Sergei)
storage/maria/ha_maria.h:
  Added prototype for start_stmt()
storage/maria/lockman.c:
  Function call rename
storage/maria/ma_bitmap.c:
  Mark deleted pages free from page cache
storage/maria/ma_blockrec.c:
  Offset -> rownr
  More debugging
  Fixed problem with small head block + long varchar
  Added logging of changed pages
  Added logging of undo (Including only loggging of changed fields in case of update)
  Added pinning/unpinning of all changed pages
  More comments
  Added free_full_pages() as the same code was used in several places.
  fill_rows_parts() renamed as fill_insert_undo_parts()
  offset -> rownr
  Added some optimization of not transactional tables
  _ma_update_block_record() has new parameter, as we need original row to do efficent undo for update
storage/maria/ma_blockrec.h:
  Added ROW_EXTENTS_ON_STACK
  Changed prototype for update and delete of row
storage/maria/ma_check.c:
  Added original row to delete_record() call
storage/maria/ma_control_file.h:
  Added ifdefs for C++
storage/maria/ma_delete.c:
  Added original row to delete_record() call
  (Needed for efficent undo logging)
storage/maria/ma_dynrec.c:
  Added extra argument to delete_record() and update_record()
  Removed not used variable
storage/maria/ma_init.c:
  Initialize log handler
storage/maria/ma_loghandler.c:
  Removed not used variable
  Change initialization of log_record_type_descriptors to not be depending on enum order
  Use array of LEX_STRING's to send data to log handler
storage/maria/ma_loghandler.h:
  New defines
  Use array of LEX_STRING's to send data to log handler
storage/maria/ma_open.c:
  Added 'dummy' transaction option to MARIA_INFO so that we can always assume 'trn' exists.
  Store in MARIA_SHARE->page_type if pages will have up to date LSN's
storage/maria/ma_pagecache.c:
  Don't decrease number of readers when using pagecache_write()/pagecache_read()
  In pagecache_write() decrement request count if page was left pinned
  Added pagecache_delete_pages()
  Removed some casts
  Make trace output consistent with rest of code
  Simplify calling of DBUG_ASSERT(0)
  Only update LSN if the LSN is bigger than what's already on the page
  Added LSN parameter pagecache_unpin_page(), pagecache_unpin(), and pagecache_unlock()
  (Part of patch from Sanja)
storage/maria/ma_static.c:
  Added 'dummy' transaction option to MARIA_INFO so that we can always assume 'trn' exists.
  Added default page cache
storage/maria/ma_statrec.c:
  Added extra argument to delete_record() and update_record()
storage/maria/ma_test1.c:
  Added option -T for transactions
storage/maria/ma_test2.c:
  Added option -T for transactions
storage/maria/ma_test_all.sh:
  Test with transactions
storage/maria/ma_update.c:
  Changed prototype for update of row
storage/maria/maria_def.h:
  Changed prototype for update & delete of row as block records need to access the old row
  Store in MARIA_SHARE->page_type if pages will have up to date LSN's
  Added MARIA_MAX_TREE_LEVELS to allow us to calculate the number of possible pinned pages we may need.
  Removed not used 'empty_bits_buffer'
  Added pointer to transaction object
  Added array for pinned pages
  Added log_row_parts array for logging of field data.
  Added MARIA_PINNED_PAGE to store pinned pages
storage/maria/trnman.c:
  Added accessor functions to transaction object
  Added missing DBUG_RETURN()
  More debugging
  More comments
  Changed // comment of code to #ifdef NOT_USED
  Transaction manager integrated.
  Post review fixes
  Part of patch originally from Sergei
storage/maria/trnman.h:
  Split trnman.h into two files, so that we don't have to include my_atomic.h into the .cc program.
  (Temporary fix to avoid bug in gcc)
storage/maria/unittest/ma_pagecache_single.c:
  Added missing argument
  Added SKIP_BIG_TESTS
  (Patch from Sanja)
storage/maria/unittest/ma_test_loghandler-t.c:
  Test logging with new LEX_STRING parameter
  (Patch from Sanja)
storage/maria/unittest/ma_test_loghandler_multigroup-t.c:
  Test logging with new LEX_STRING parameter
  (Patch from Sanja)
storage/maria/unittest/ma_test_loghandler_multithread-t.c:
  Test logging with new LEX_STRING parameter
  (Patch from Sanja)
storage/maria/unittest/ma_test_loghandler_pagecache-t.c:
  Test logging with new LEX_STRING parameter
  (Patch from Sanja)
storage/maria/unittest/trnman-t.c:
  Stack overflow detection
  (Patch from Sergei)
unittest/unit.pl:
  Command-line options --big and --verbose
  (Patch from Sergei)
unittest/mytap/tap.c:
  Detect --big
  (Patch from Sergei)
unittest/mytap/tap.h:
  Skip_big_tests and SKIP_BIG_TESTS
  (Patch from Sergei)
storage/maria/trnman_public.h:
  New BitKeeper file ``storage/maria/trnman_public.h''

include/lf.h

-/* Copyright (C) 2006 MySQL AB
+/* 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
@@ -96,26 +96,28 @@ nolock_wrap(lf_dynarray_iterate, int,
 #define LF_PINBOX_PINS 4
 #define LF_PURGATORY_SIZE 10
 
-typedef void lf_pinbox_free_func(void *, void *);
+typedef void lf_pinbox_free_func(void *, void *, void*);
 
 typedef struct {
-  LF_DYNARRAY pinstack;
+  LF_DYNARRAY pinarray;
   lf_pinbox_free_func *free_func;
   void *free_func_arg;
   uint free_ptr_offset;
   uint32 volatile pinstack_top_ver;         /* this is a versioned pointer */
-  uint32 volatile pins_in_stack;            /* number of elements in array */
+  uint32 volatile pins_in_array;            /* number of elements in array */
 } LF_PINBOX;
 
 typedef struct {
   void * volatile pin[LF_PINBOX_PINS];
   LF_PINBOX *pinbox;
+  void  *stack_ends_here;
   void  *purgatory;
   uint32 purgatory_count;
   uint32 volatile link;
 /* we want sizeof(LF_PINS) to be 128 to avoid false sharing */
   char pad[128-sizeof(uint32)*2
               -sizeof(LF_PINBOX *)
+              -sizeof(void*)
               -sizeof(void *)*(LF_PINBOX_PINS+1)];
 } LF_PINS;
 
@@ -124,9 +126,9 @@ typedef struct {
   (e.g. lf_hash).
 */
 #define lf_rwlock_by_pins(PINS)   \
-  my_atomic_rwlock_wrlock(&(PINS)->pinbox->pinstack.lock)
+  my_atomic_rwlock_wrlock(&(PINS)->pinbox->pinarray.lock)
 #define lf_rwunlock_by_pins(PINS) \
-  my_atomic_rwlock_wrunlock(&(PINS)->pinbox->pinstack.lock)
+  my_atomic_rwlock_wrunlock(&(PINS)->pinbox->pinarray.lock)
 
 /*
   compile-time assert, to require "no less than N" pins
@@ -135,9 +137,9 @@ typedef struct {
 */
 #if defined(__GNUC__) && defined(MY_LF_EXTRA_DEBUG)
 #define LF_REQUIRE_PINS(N)                                      \
-  static const char require_pins[LF_PINBOX_PINS-N] __attribute__ ((unused)); \
-  static const int LF_NUM_PINS_IN_THIS_FILE= N
-
+  static const char require_pins[LF_PINBOX_PINS-N]              \
+                             __attribute__ ((unused));          \
+  static const int LF_NUM_PINS_IN_THIS_FILE= N;
 #define _lf_pin(PINS, PIN, ADDR)                                \
   (                                                             \
     assert(PIN < LF_NUM_PINS_IN_THIS_FILE),                     \
@@ -164,17 +166,17 @@ void lf_pinbox_init(LF_PINBOX *pinbox, uint free_ptr_offset,
 void lf_pinbox_destroy(LF_PINBOX *pinbox);
 
 lock_wrap(lf_pinbox_get_pins, LF_PINS *,
-          (LF_PINBOX *pinbox),
-          (pinbox),
-          &pinbox->pinstack.lock)
+          (LF_PINBOX *pinbox, void *stack_end),
+          (pinbox, stack_end),
+          &pinbox->pinarray.lock);
 lock_wrap_void(lf_pinbox_put_pins,
                (LF_PINS *pins),
                (pins),
-               &pins->pinbox->pinstack.lock)
+               &pins->pinbox->pinarray.lock);
 lock_wrap_void(lf_pinbox_free,
                (LF_PINS *pins, void *addr),
                (pins, addr),
-               &pins->pinbox->pinstack.lock)
+               &pins->pinbox->pinarray.lock);
 
 /*
   memory allocator, lf_alloc-pin.c
@@ -193,23 +195,23 @@ typedef struct st_lf_allocator {
 
 void lf_alloc_init(LF_ALLOCATOR *allocator, uint size, uint free_ptr_offset);
 void lf_alloc_destroy(LF_ALLOCATOR *allocator);
-uint lf_alloc_in_pool(LF_ALLOCATOR *allocator);
+uint lf_alloc_pool_count(LF_ALLOCATOR *allocator);
 /*
   shortcut macros to access underlying pinbox functions from an LF_ALLOCATOR
   see _lf_pinbox_get_pins() and _lf_pinbox_put_pins()
 */
-#define _lf_alloc_free(PINS, PTR)   _lf_pinbox_free((PINS), (PTR))
-#define lf_alloc_free(PINS, PTR)    lf_pinbox_free((PINS), (PTR))
-#define _lf_alloc_get_pins(ALLOC)   _lf_pinbox_get_pins(&(ALLOC)->pinbox)
-#define lf_alloc_get_pins(ALLOC)    lf_pinbox_get_pins(&(ALLOC)->pinbox)
-#define _lf_alloc_put_pins(PINS)    _lf_pinbox_put_pins(PINS)
-#define lf_alloc_put_pins(PINS)     lf_pinbox_put_pins(PINS)
-#define lf_alloc_real_free(ALLOC, ADDR) my_free((gptr)(ADDR), MYF(0))
+#define _lf_alloc_free(PINS, PTR)     _lf_pinbox_free((PINS), (PTR))
+#define lf_alloc_free(PINS, PTR)       lf_pinbox_free((PINS), (PTR))
+#define _lf_alloc_get_pins(A, ST)     _lf_pinbox_get_pins(&(A)->pinbox, (ST))
+#define lf_alloc_get_pins(A, ST)       lf_pinbox_get_pins(&(A)->pinbox, (ST))
+#define _lf_alloc_put_pins(PINS)      _lf_pinbox_put_pins(PINS)
+#define lf_alloc_put_pins(PINS)        lf_pinbox_put_pins(PINS)
+#define lf_alloc_direct_free(ALLOC, ADDR) my_free((gptr)(ADDR), MYF(0))
 
 lock_wrap(lf_alloc_new, void *,
           (LF_PINS *pins),
           (pins),
-          &pins->pinbox->pinstack.lock)
+          &pins->pinbox->pinarray.lock);
 
 /*
   extendible hash, lf_hash.c
@@ -240,11 +242,11 @@ int lf_hash_delete(LF_HASH *hash, LF_PINS *pins, const void *key, uint keylen);
   shortcut macros to access underlying pinbox functions from an LF_HASH
   see _lf_pinbox_get_pins() and _lf_pinbox_put_pins()
 */
-#define _lf_hash_get_pins(HASH)   _lf_alloc_get_pins(&(HASH)->alloc)
-#define lf_hash_get_pins(HASH)    lf_alloc_get_pins(&(HASH)->alloc)
-#define _lf_hash_put_pins(PINS)   _lf_pinbox_put_pins(PINS)
-#define lf_hash_put_pins(PINS)    lf_pinbox_put_pins(PINS)
-
+#define _lf_hash_get_pins(HASH, ST) _lf_alloc_get_pins(&(HASH)->alloc, (ST))
+#define lf_hash_get_pins(HASH, ST)   lf_alloc_get_pins(&(HASH)->alloc, (ST))
+#define _lf_hash_put_pins(PINS)     _lf_pinbox_put_pins(PINS)
+#define lf_hash_put_pins(PINS)       lf_pinbox_put_pins(PINS)
+#define lf_hash_search_unpin(PINS)   lf_unpin((PINS), 2)
 /*
   cleanup
 */

include/my_atomic.h

@@ -225,6 +225,11 @@ make_atomic_fas(ptr)
 #undef make_atomic_fas_body
 #undef intptr
 
+/*
+  the macro below defines (as an expression) the code that
+  will be run in spin-loops. Intel manuals recummend to have PAUSE there.
+  It is expected to be defined in include/atomic/ *.h files
+*/
 #ifndef LF_BACKOFF
 #define LF_BACKOFF (1)
 #endif

include/my_global.h

@@ -1034,6 +1034,8 @@ typedef long intptr;
 #error
 #endif
 
+#define MY_ERRPTR ((void*)(intptr)1)
+
 #ifdef USE_RAID
 /*
   The following is done with a if to not get problems with pre-processors

include/pagecache.h

@@ -206,17 +206,21 @@ extern void pagecache_unlock_page(PAGECACHE *pagecache,
                                   pgcache_page_no_t pageno,
                                   enum pagecache_page_lock lock,
                                   enum pagecache_page_pin pin,
-                                  LSN first_REDO_LSN_for_page);
+                                  LSN first_REDO_LSN_for_page,
+                                  LSN lsn);
 extern void pagecache_unlock(PAGECACHE *pagecache,
                              PAGECACHE_PAGE_LINK *link,
                              enum pagecache_page_lock lock,
                              enum pagecache_page_pin pin,
-                             LSN first_REDO_LSN_for_page);
+                             LSN first_REDO_LSN_for_page,
+                             LSN lsn);
 extern void pagecache_unpin_page(PAGECACHE *pagecache,
                                  PAGECACHE_FILE *file,
-                                 pgcache_page_no_t pageno);
+                                 pgcache_page_no_t pageno,
+                                 LSN lsn);
 extern void pagecache_unpin(PAGECACHE *pagecache,
-                            PAGECACHE_PAGE_LINK *link);
+                            PAGECACHE_PAGE_LINK *link,
+                            LSN lsn);
 extern int flush_pagecache_blocks(PAGECACHE *keycache,
                                   PAGECACHE_FILE *file,
                                   enum flush_type type);
@@ -225,6 +229,12 @@ extern my_bool pagecache_delete_page(PAGECACHE *pagecache,
                                      pgcache_page_no_t pageno,
                                      enum pagecache_page_lock lock,
                                      my_bool flush);
+extern my_bool pagecache_delete_pages(PAGECACHE *pagecache,
+                                      PAGECACHE_FILE *file,
+                                      pgcache_page_no_t pageno,
+                                      uint page_count,
+                                      enum pagecache_page_lock lock,
+                                      my_bool flush);
 extern void end_pagecache(PAGECACHE *keycache, my_bool cleanup);
 extern my_bool pagecache_collect_changed_blocks_with_lsn(PAGECACHE *pagecache,
                                                          LEX_STRING *str,

mysql-test/r/maria.result

@@ -78,10 +78,15 @@ explain select a,b,c from t1;
 id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
 1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	4	
 drop table t1;
+set autocommit=0;
+begin;
 CREATE TABLE t1 (a INT);
 INSERT INTO  t1 VALUES (1), (2), (3);
 LOCK TABLES t1 WRITE;
 INSERT INTO  t1 VALUES (1), (2), (3);
+commit;
+set autocommit=1;
+UNLOCK TABLES;
 OPTIMIZE TABLE t1;
 Table	Op	Msg_type	Msg_text
 test.t1	optimize	status	OK
@@ -522,12 +527,15 @@ create table t1 (a int not null, primary key(a)) ROW_FORMAT=FIXED;
 create table t2 (a int not null, b int not null, primary key(a,b)) ROW_FORMAT=FIXED;
 insert into t1 values (1),(2),(3),(4),(5),(6);
 insert into t2 values (1,1),(2,1);
+set autocommit=0;
+begin;
 lock tables t1 read local, t2 read local;
 select straight_join * from t1,t2 force index (primary) where t1.a=t2.a;
 a	a	b
 1	1	1
 2	2	1
 insert into t2 values(2,0);
+commit;
 select straight_join * from t1,t2 force index (primary) where t1.a=t2.a;
 a	a	b
 1	1	1

mysql-test/t/maria.test

@@ -90,10 +90,15 @@ drop table t1;
 #
 # Test of OPTIMIZE of locked and modified tables
 #
+set autocommit=0;
+begin;
 CREATE TABLE t1 (a INT);
 INSERT INTO  t1 VALUES (1), (2), (3);
 LOCK TABLES t1 WRITE;
 INSERT INTO  t1 VALUES (1), (2), (3);
+commit;
+set autocommit=1;
+UNLOCK TABLES;
 OPTIMIZE TABLE t1;
 DROP TABLE t1;
 
@@ -510,10 +515,13 @@ create table t1 (a int not null, primary key(a)) ROW_FORMAT=FIXED;
 create table t2 (a int not null, b int not null, primary key(a,b)) ROW_FORMAT=FIXED;
 insert into t1 values (1),(2),(3),(4),(5),(6);
 insert into t2 values (1,1),(2,1);
+set autocommit=0;
+begin;
 lock tables t1 read local, t2 read local;
 select straight_join * from t1,t2 force index (primary) where t1.a=t2.a;
 connect (root,localhost,root,,test,$MASTER_MYPORT,$MASTER_MYSOCK);
 insert into t2 values(2,0);
+commit;
 disconnect root;
 connection default;
 select straight_join * from t1,t2 force index (primary) where t1.a=t2.a;

mysys/lf_dynarray.c

-/* Copyright (C) 2000 MySQL AB
+/* Copyright (C) 2006 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
@@ -21,8 +21,6 @@
   Memory is allocated in non-contiguous chunks.
   This data structure is not space efficient for sparse arrays.
 
-  The number of elements is limited to 4311810304
-
   Every element is aligned to sizeof(element) boundary
   (to avoid false sharing if element is big enough).
 
@@ -32,6 +30,9 @@
   to arrays of elements, on the second level it's an array of pointers
   to arrays of pointers to arrays of elements. And so on.
 
+  With four levels the number of elements is limited to 4311810304
+  (but as in all functions index is uint, the real limit is 2^32-1)
+
   Actually, it's wait-free, not lock-free ;-)
 */
 
@@ -192,6 +193,9 @@ static int recursive_iterate(LF_DYNARRAY *array, void *ptr, int level,
     each. _lf_dynarray_iterate() calls user-supplied function on every array
     from the set. It is the fastest way to scan the array, faster than
       for (i=0; i < N; i++) { func(_lf_dynarray_value(dynarray, i)); }
+
+  NOTE
+    if func() returns non-zero, the scan is aborted
 */
 int _lf_dynarray_iterate(LF_DYNARRAY *array, lf_dynarray_func func, void *arg)
 {

storage/maria/ha_maria.cc

@@ -25,12 +25,14 @@
 #include <myisampack.h>
 #include <my_bit.h>
 #include "ha_maria.h"
+#include "trnman_public.h"
 
 #include "maria_def.h"
 #include "ma_rt_index.h"
 #include "ma_blockrec.h"
 
 ulong maria_recover_options= HA_RECOVER_NONE;
+static handlerton *maria_hton;
 
 /* bits in maria_recover_options */
 const char *maria_recover_names[]=
@@ -464,7 +466,7 @@ ha_maria::ha_maria(handlerton *hton, TABLE_SHARE *table_arg):
 handler(hton, table_arg), file(0),
 int_table_flags(HA_NULL_IN_KEY | HA_CAN_FULLTEXT | HA_CAN_SQL_HANDLER |
                 HA_DUPLICATE_POS | HA_CAN_INDEX_BLOBS | HA_AUTO_PART_KEY |
-                HA_FILE_BASED | HA_CAN_GEOMETRY | HA_NO_TRANSACTIONS |
+                HA_FILE_BASED | HA_CAN_GEOMETRY |
                 HA_CAN_INSERT_DELAYED | HA_CAN_BIT_FIELD | HA_CAN_RTREEKEYS |
                 HA_HAS_RECORDS | HA_STATS_RECORDS_IS_EXACT),
 can_enable_indexes(1)
@@ -1846,14 +1848,69 @@ int ha_maria::delete_table(const char *name)
   return maria_delete_table(name);
 }
 
+#define THD_TRN (*(TRN **)thd_ha_data(thd, maria_hton))
 
 int ha_maria::external_lock(THD *thd, int lock_type)
 {
-  return maria_lock_database(file, !table->s->tmp_table ?
-                             lock_type : ((lock_type == F_UNLCK) ?
-                                          F_UNLCK : F_EXTRA_LCK));
+  TRN *trn= THD_TRN;
+  DBUG_ENTER("ha_maria::external_lock");
+  if (!trn && lock_type != F_UNLCK) /* no transaction yet - open it now */
+  {
+    trn= trnman_new_trn(& thd->mysys_var->mutex,
+                        & thd->mysys_var->suspend,
+                        thd->thread_stack + STACK_DIRECTION *
+                        (my_thread_stack_size - STACK_MIN_SIZE));
+    if (!trn)
+      DBUG_RETURN(HA_ERR_OUT_OF_MEM);
+
+    DBUG_PRINT("info", ("THD_TRN set to 0x%lx", (ulong)trn));
+    THD_TRN= trn;
+    if (thd->options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN))
+      trans_register_ha(thd, true, maria_hton);
+  }
+  if (lock_type != F_UNLCK)
+  {
+    this->file->trn= trn;
+    if (!trnman_increment_locked_tables(trn))
+    {
+      trans_register_ha(thd, FALSE, maria_hton);
+      trnman_new_statement(trn);
+    }
+  }
+  else
+  {
+    this->file->trn= 0; /* TODO: remove it also in commit and rollback */
+    if (trn && trnman_has_locked_tables(trn))
+    {
+      if (!trnman_decrement_locked_tables(trn))
+      {
+        /* autocommit ? rollback a transaction */
+        if (!(thd->options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN)))
+        {
+          trnman_rollback_trn(trn);
+          DBUG_PRINT("info", ("THD_TRN set to 0x0"));
+          THD_TRN= 0;
+        }
+      }
+    }
+  }
+  DBUG_RETURN(maria_lock_database(file, !table->s->tmp_table ?
+                                  lock_type : ((lock_type == F_UNLCK) ?
+                                               F_UNLCK : F_EXTRA_LCK)));
 }
 
+int ha_maria::start_stmt(THD *thd, thr_lock_type lock_type)
+{
+  TRN *trn= THD_TRN;
+  DBUG_ASSERT(trn); // this may be called only after external_lock()
+  DBUG_ASSERT(lock_type != F_UNLCK);
+  if (!trnman_increment_locked_tables(trn))
+  {
+    trans_register_ha(thd, false, maria_hton);
+    trnman_new_statement(trn);
+  }
+  return 0;
+}
 
 THR_LOCK_DATA **ha_maria::store_lock(THD *thd,
                                      THR_LOCK_DATA **to,
@@ -1936,6 +1993,7 @@ int ha_maria::create(const char *name, register TABLE *table_arg,
                                  share->avg_row_length);
   create_info.data_file_name= ha_create_info->data_file_name;
   create_info.index_file_name= ha_create_info->index_file_name;
+  create_info.transactional= row_type == BLOCK_RECORD;
 
   if (ha_create_info->options & HA_LEX_CREATE_TMP_TABLE)
     create_flags|= HA_CREATE_TMP_TABLE;
@@ -2089,26 +2147,72 @@ bool ha_maria::check_if_incompatible_data(HA_CREATE_INFO *info,
   return COMPATIBLE_DATA_YES;
 }
 
-extern int maria_panic(enum ha_panic_function flag);
-int maria_panic(handlerton *hton, ha_panic_function flag)
+
+static int maria_hton_panic(handlerton *hton, ha_panic_function flag)
 {
   return maria_panic(flag);
 }
 
-static int ha_maria_init(void *p)
+
+static int maria_commit(handlerton *hton __attribute__ ((unused)),
+                        THD *thd, bool all)
 {
-  handlerton *maria_hton;
+  TRN *trn= THD_TRN;
+  DBUG_ENTER("maria_commit");
+  trnman_reset_locked_tables(trn);
+  /* statement or transaction ? */
+  if ((thd->options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN)) && !all)
+    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
+}
 
+
+static int maria_rollback(handlerton *hton __attribute__ ((unused)),
+                          THD *thd, bool all)
+{
+  TRN *trn= THD_TRN;
+  DBUG_ENTER("maria_rollback");
+  trnman_reset_locked_tables(trn);
+  /* statement or transaction ? */
+  if ((thd->options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN)) && !all)
+  {
+    trnman_rollback_statement(trn);
+    DBUG_RETURN(0); // end of statement
+  }
+  DBUG_PRINT("info", ("THD_TRN set to 0x0"));
+  THD_TRN= 0;
+  DBUG_RETURN(trnman_rollback_trn(trn) ?
+              HA_ERR_OUT_OF_MEM : 0); // end of transaction
+}
+
+
+static int ha_maria_init(void *p)
+{
   maria_hton= (handlerton *)p;
   maria_hton->state= SHOW_OPTION_YES;
   maria_hton->db_type= DB_TYPE_MARIA;
   maria_hton->create= maria_create_handler;
-  maria_hton->panic= maria_panic;
+  maria_hton->panic= maria_hton_panic;
+  maria_hton->commit= maria_commit;
+  maria_hton->rollback= maria_rollback;
   /* TODO: decide if we support Maria being used for log tables */
   maria_hton->flags= HTON_CAN_RECREATE | HTON_SUPPORT_LOG_TABLES;
-  return test(maria_init());
+  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()));
 }
 
+
 struct st_mysql_storage_engine maria_storage_engine=
 { MYSQL_HANDLERTON_INTERFACE_VERSION };
 

storage/maria/ha_maria.h

@@ -110,6 +110,7 @@ class ha_maria :public handler
   int extra_opt(enum ha_extra_function operation, ulong cache_size);
   int reset(void);
   int external_lock(THD * thd, int lock_type);
+  int start_stmt(THD *thd, thr_lock_type lock_type);
   int delete_all_rows(void);
   int disable_indexes(uint mode);
   int enable_indexes(uint mode);

storage/maria/lockman.c

@@ -538,7 +538,7 @@ void lockman_destroy(LOCKMAN *lm)
   {
     intptr next= el->link;
     if (el->hashnr & 1)
-      lf_alloc_real_free(&lm->alloc, el);
+      lf_alloc_direct_free(&lm->alloc, el);
     else
       my_free((void *)el, MYF(0));
     el= (LOCK *)next;

storage/maria/ma_bitmap.c

@@ -1837,7 +1837,7 @@ my_bool _ma_bitmap_release_unused(MARIA_HA *info, MARIA_BITMAP_BLOCKS *blocks)
 
 
 /*
-  Free full pages from bitmap
+  Free full pages from bitmap and pagecache
 
   SYNOPSIS
     _ma_bitmap_free_full_pages()
@@ -1846,7 +1846,8 @@ my_bool _ma_bitmap_release_unused(MARIA_HA *info, MARIA_BITMAP_BLOCKS *blocks)
     count               Number of extents
 
   IMPLEMENTATION
-    Mark all full pages (not tails) from extents as free
+    Mark all full pages (not tails) from extents as free, both in bitmap
+    and page cache.
 
   RETURN
     0  ok
@@ -1865,6 +1866,9 @@ my_bool _ma_bitmap_free_full_pages(MARIA_HA *info, const byte *extents,
     uint page_count= uint2korr(extents + ROW_EXTENT_PAGE_SIZE);
     if (!(page_count & TAIL_BIT))
     {
+      if (pagecache_delete_pages(info->s->pagecache, &info->dfile, page,
+                                 page_count, PAGECACHE_LOCK_WRITE, 1))
+        DBUG_RETURN(1);
       if (_ma_reset_full_page_bits(info, &info->s->bitmap, page, page_count))
       {
         pthread_mutex_unlock(&info->s->bitmap.bitmap_lock);

storage/maria/ma_blockrec.h

@@ -102,6 +102,9 @@ enum en_page_type { UNALLOCATED_PAGE, HEAD_PAGE, TAIL_PAGE, BLOB_PAGE, MAX_PAGE_
 */
 #define MAX_TAIL_SIZE(block_size) ((block_size) *3 / 4)
 
+/* Don't allocate memory for too many row extents on the stack */
+#define ROW_EXTENTS_ON_STACK	32
+
 extern uchar maria_bitmap_marker[2];
 
 /* Functions to convert MARIA_RECORD_POS to/from page:offset */
@@ -130,8 +133,8 @@ my_bool _ma_init_block_record(MARIA_HA *info);
 void _ma_end_block_record(MARIA_HA *info);
 
 my_bool _ma_update_block_record(MARIA_HA *info, MARIA_RECORD_POS pos,
-                                const byte *record);
-my_bool _ma_delete_block_record(MARIA_HA *info);
+                                const byte *oldrec, const byte *newrec);
+my_bool _ma_delete_block_record(MARIA_HA *info, const byte *record);
 int     _ma_read_block_record(MARIA_HA *info, byte *record,
                               MARIA_RECORD_POS record_pos);
 int _ma_read_block_record2(MARIA_HA *info, byte *record,

storage/maria/ma_check.c

@@ -4507,7 +4507,8 @@ static int sort_delete_record(MARIA_SORT_PARAM *sort_param)
     if (sort_param->calc_checksum)
       param->glob_crc-=(*info->s->calc_checksum)(info, sort_param->record);
   }
-  error=flush_io_cache(&info->rec_cache) || (*info->s->delete_record)(info);
+  error= (flush_io_cache(&info->rec_cache) ||
+          (*info->s->delete_record)(info, sort_param->record));
   info->dfile.file= old_file;			/* restore actual value */
   info->state->records--;
   DBUG_RETURN(error);

storage/maria/ma_control_file.h

@@ -56,6 +56,9 @@ typedef enum enum_control_file_error {
 #define CONTROL_FILE_UPDATE_ONLY_LSN 1
 #define CONTROL_FILE_UPDATE_ONLY_LOGNO 2
 
+#ifdef	__cplusplus
+extern "C" {
+#endif
 
 /*
   Looks for the control file. If absent, it's a fresh start, create file.
@@ -74,3 +77,7 @@ int ma_control_file_write_and_force(const LSN checkpoint_lsn, uint32 logno,
 
 /* Free resources taken by control file subsystem */
 int ma_control_file_end();
+
+#ifdef	__cplusplus
+}
+#endif

storage/maria/ma_delete.c

@@ -89,7 +89,7 @@ int maria_delete(MARIA_HA *info,const byte *record)
     }
   }
 
-  if ((*share->delete_record)(info))
+  if ((*share->delete_record)(info, record))
     goto err;				/* Remove record from database */
 
   /*

storage/maria/ma_dynrec.c

@@ -237,6 +237,7 @@ my_bool _ma_write_dynamic_record(MARIA_HA *info, const byte *record)
 }
 
 my_bool _ma_update_dynamic_record(MARIA_HA *info, MARIA_RECORD_POS pos,
+                                  const byte *oldrec __attribute__ ((unused)),
                                   const byte *record)
 {
   uint length= _ma_rec_pack(info, info->rec_buff + MARIA_REC_BUFF_OFFSET,
@@ -277,6 +278,7 @@ my_bool _ma_write_blob_record(MARIA_HA *info, const byte *record)
 
 
 my_bool _ma_update_blob_record(MARIA_HA *info, MARIA_RECORD_POS pos,
+                               const byte *oldrec __attribute__ ((unused)),
                                const byte *record)
 {
   byte *rec_buff;
@@ -309,7 +311,8 @@ my_bool _ma_update_blob_record(MARIA_HA *info, MARIA_RECORD_POS pos,
 }
 
 
-my_bool _ma_delete_dynamic_record(MARIA_HA *info)
+my_bool _ma_delete_dynamic_record(MARIA_HA *info,
+                                  const byte *record __attribute__ ((unused)))
 {
   return delete_dynamic_record(info, info->cur_row.lastpos, 0);
 }
@@ -1371,7 +1374,7 @@ int _ma_read_dynamic_record(MARIA_HA *info, byte *buf,
                             MARIA_RECORD_POS filepos)
 {
   int block_of_record;
-  uint b_type,left_length;
+  uint b_type;
   MARIA_BLOCK_INFO block_info;
   File file;
   DBUG_ENTER("_ma_read_dynamic_record");

storage/maria/ma_init.c

@@ -43,6 +43,7 @@ int maria_init(void)
     maria_inited= TRUE;
     pthread_mutex_init(&THR_LOCK_maria,MY_MUTEX_INIT_SLOW);
     _ma_init_block_record_data();
+    loghandler_init();
   }
   return 0;
 }

storage/maria/ma_loghandler.h

+/* transaction log default cache size  (TODO: make it global variable) */
+#define TRANSLOG_PAGECACHE_SIZE 1024*1024*2
+/* transaction log default file size  (TODO: make it global variable) */
+#define TRANSLOG_FILE_SIZE 1024*1024*1024
+/* transaction log default flags (TODO: make it global variable) */
+#define TRANSLOG_DEFAULT_FLAGS 0
 
 /* Transaction log flags */
 #define TRANSLOG_PAGE_CRC              1
@@ -18,48 +24,77 @@
 /* short transaction ID type */
 typedef uint16 SHORT_TRANSACTION_ID;
 
+struct st_maria_share;
+
 /* Length of CRC at end of pages */
 #define CRC_LENGTH 4
+/* Size of file id in logs */
+#define FILEID_STORE_SIZE 2
+/* Size of page reference in log */
+#define PAGE_STORE_SIZE ROW_EXTENT_PAGE_SIZE
+/* Size of page ranges in log */
+#define PAGERANGE_STORE_SIZE ROW_EXTENT_COUNT_SIZE
+#define DIRPOS_STORE_SIZE 1
+
+/* Store methods to match the above sizes */
+#define fileid_store(T,A) int2store(T,A)
+#define page_store(T,A)   int5store(T,A)
+#define dirpos_store(T,A) ((*(uchar*) (T)) = A)
+#define pagerange_store(T,A) int2store(T,A)
+
 /*
   Length of disk drive sector size (we assume that writing it
   to disk is atomic operation)
 */
 #define DISK_DRIVE_SECTOR_SIZE 512
 
+/*
+  Number of empty entries we need to have in LEX_STRING for
+  translog_write_record()
+*/
+#define LOG_INTERNAL_PARTS 1
+
+/* position reserved in an array of parts of a log record */
+#define TRANSLOG_INTERNAL_PARTS 1
+
 /* types of records in the transaction log */
+/* Todo: Set numbers for these when we have all entries figured out */
+
 enum translog_record_type
 {
   LOGREC_RESERVED_FOR_CHUNKS23= 0,
-  LOGREC_REDO_INSERT_ROW_HEAD= 1,
-  LOGREC_REDO_INSERT_ROW_TAIL= 2,
-  LOGREC_REDO_INSERT_ROW_BLOB= 3,
-  LOGREC_REDO_INSERT_ROW_BLOBS= 4,
-  LOGREC_REDO_PURGE_ROW= 5,
-  eLOGREC_REDO_PURGE_BLOCKS= 6,
-  LOGREC_REDO_DELETE_ROW= 7,
-  LOGREC_REDO_UPDATE_ROW_HEAD= 8,
-  LOGREC_REDO_INDEX= 9,
-  LOGREC_REDO_UNDELETE_ROW= 10,
-  LOGREC_CLR_END= 11,
-  LOGREC_PURGE_END= 12,
-  LOGREC_UNDO_ROW_INSERT= 13,
-  LOGREC_UNDO_ROW_DELETE= 14,
-  LOGREC_UNDO_ROW_UPDATE= 15,
-  LOGREC_UNDO_KEY_INSERT= 16,
-  LOGREC_UNDO_KEY_DELETE= 17,
-  LOGREC_PREPARE= 18,
-  LOGREC_PREPARE_WITH_UNDO_PURGE= 19,
-  LOGREC_COMMIT= 20,
-  LOGREC_COMMIT_WITH_UNDO_PURGE= 21,
-  LOGREC_CHECKPOINT_PAGE= 22,
-  LOGREC_CHECKPOINT_TRAN= 23,
-  LOGREC_CHECKPOINT_TABL= 24,
-  LOGREC_REDO_CREATE_TABLE= 25,
-  LOGREC_REDO_RENAME_TABLE= 26,
-  LOGREC_REDO_DROP_TABLE= 27,
-  LOGREC_REDO_TRUNCATE_TABLE= 28,
-  LOGREC_FILE_ID= 29,
-  LOGREC_LONG_TRANSACTION_ID= 30,
+  LOGREC_REDO_INSERT_ROW_HEAD,
+  LOGREC_REDO_INSERT_ROW_TAIL,
+  LOGREC_REDO_INSERT_ROW_BLOB,
+  LOGREC_REDO_INSERT_ROW_BLOBS,
+  LOGREC_REDO_PURGE_ROW_HEAD,
+  LOGREC_REDO_PURGE_ROW_TAIL,
+  LOGREC_REDO_PURGE_BLOCKS,
+  LOGREC_REDO_DELETE_ROW,
+  LOGREC_REDO_UPDATE_ROW_HEAD,
+  LOGREC_REDO_INDEX,
+  LOGREC_REDO_UNDELETE_ROW,
+  LOGREC_CLR_END,
+  LOGREC_PURGE_END,
+  LOGREC_UNDO_ROW_INSERT,
+  LOGREC_UNDO_ROW_DELETE,
+  LOGREC_UNDO_ROW_UPDATE,
+  LOGREC_UNDO_ROW_PURGE,
+  LOGREC_UNDO_KEY_INSERT,
+  LOGREC_UNDO_KEY_DELETE,
+  LOGREC_PREPARE,
+  LOGREC_PREPARE_WITH_UNDO_PURGE,
+  LOGREC_COMMIT,
+  LOGREC_COMMIT_WITH_UNDO_PURGE,
+  LOGREC_CHECKPOINT_PAGE,
+  LOGREC_CHECKPOINT_TRAN,
+  LOGREC_CHECKPOINT_TABL,
+  LOGREC_REDO_CREATE_TABLE,
+  LOGREC_REDO_RENAME_TABLE,
+  LOGREC_REDO_DROP_TABLE,
+  LOGREC_REDO_TRUNCATE_TABLE,
+  LOGREC_FILE_ID,
+  LOGREC_LONG_TRANSACTION_ID,
   LOGREC_RESERVED_FUTURE_EXTENSION= 63
 };
 #define LOGREC_NUMBER_OF_TYPES 64              /* Maximum, can't be extended */
@@ -145,17 +180,22 @@ struct st_translog_reader_data
   my_bool eor;                                  /* end of the record */
 };
 
+#ifdef	__cplusplus
+extern "C" {
+#endif
 
+extern void loghandler_init();
 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,
-				     SHORT_TRANSACTION_ID short_trid,
-				     void *tcb,
-				     translog_size_t part1_length,
-				     byte *part1_buff, ...);
+                                     enum translog_record_type type,
+                                     SHORT_TRANSACTION_ID short_trid,
+                                     void *tcb, struct st_maria_share *share,
+                                     translog_size_t rec_len,
+                                     uint part_no,
+                                     LEX_STRING *parts_data);
 
 extern void translog_destroy();
 
@@ -182,4 +222,7 @@ extern translog_size_t translog_read_next_record_header(TRANSLOG_SCANNER_DATA
 							*scanner,
 							TRANSLOG_HEADER_BUFFER
 							*buff);
+#ifdef	__cplusplus
+}
+#endif
 

storage/maria/ma_open.c

@@ -19,6 +19,7 @@
 #include "ma_sp_defs.h"
 #include "ma_rt_index.h"
 #include "ma_blockrec.h"
+#include "trnman.h"
 #include <m_ctype.h>
 
 #if defined(MSDOS) || defined(__WIN__)
@@ -431,6 +432,9 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
       share->base_length+= TRANS_ROW_EXTRA_HEADER_SIZE;
     share->base.default_rec_buff_size= max(share->base.pack_reclength,
                                            share->base.max_key_length);
+    share->page_type= (share->base.transactional ? PAGECACHE_LSN_PAGE :
+                       PAGECACHE_PLAIN_PAGE);
+
     if (share->data_file_type == DYNAMIC_RECORD)
     {
       share->base.extra_rec_buff_size=
@@ -634,6 +638,7 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
     share->delay_key_write=1;
 
   info.state= &share->state.state;	/* Change global values by default */
+  info.trn=   &dummy_transaction_object;
   pthread_mutex_unlock(&share->intern_lock);
 
   /* Allocate buffer for one record */

storage/maria/ma_static.c

@@ -20,6 +20,7 @@
 
 #ifndef _global_h
 #include "maria_def.h"
+#include "trnman.h"
 #endif
 
 LIST	*maria_open_list=0;
@@ -43,6 +44,12 @@ ulong    maria_data_pointer_size= 4;
 PAGECACHE maria_pagecache_var;
 PAGECACHE *maria_pagecache= &maria_pagecache_var;
 
+PAGECACHE maria_log_pagecache_var;
+PAGECACHE *maria_log_pagecache= &maria_log_pagecache_var;
+
+/* For using maria externally */
+TRN dummy_transaction_object;
+
 /* Enough for comparing if number is zero */
 byte maria_zero_string[]= {0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0};
 

storage/maria/ma_statrec.c

@@ -86,6 +86,7 @@ my_bool _ma_write_static_record(MARIA_HA *info, const byte *record)
 }
 
 my_bool _ma_update_static_record(MARIA_HA *info, MARIA_RECORD_POS pos,
+                                 const byte *oldrec __attribute__ ((unused)),
                                  const byte *record)
 {
   info->rec_cache.seek_not_done=1;		/* We have done a seek */
@@ -96,7 +97,8 @@ my_bool _ma_update_static_record(MARIA_HA *info, MARIA_RECORD_POS pos,
 }
 
 
-my_bool _ma_delete_static_record(MARIA_HA *info)
+my_bool _ma_delete_static_record(MARIA_HA *info,
+                                 const byte *record __attribute__ ((unused)))
 {
   byte temp[9];                                 /* 1+sizeof(uint32) */
   info->state->del++;

storage/maria/ma_test1.c

@@ -33,7 +33,7 @@ static uint insert_count, update_count, remove_count;
 static uint pack_keys=0, pack_seg=0, key_length;
 static uint unique_key=HA_NOSAME;
 static my_bool pagecacheing, null_fields, silent, skip_update, opt_unique,
-  verbose, skip_delete;
+  verbose, skip_delete, transactional;
 static MARIA_COLUMNDEF recinfo[4];
 static MARIA_KEYDEF keyinfo[10];
 static HA_KEYSEG keyseg[10];
@@ -152,6 +152,7 @@ static int run_test(const char *filename)
   create_info.max_rows=(ulong) (rec_pointer_size ?
 				(1L << (rec_pointer_size*8))/40 :
 				0);
+  create_info.transactional= transactional;
   if (maria_create(filename, record_type, 1, keyinfo,2+opt_unique,recinfo,
 		uniques, &uniquedef, &create_info,
 		create_flag))
@@ -595,6 +596,9 @@ static struct my_option my_long_options[] =
    (gptr*) &skip_delete, 0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
   {"skip-update", 'D', "Don't test updates", (gptr*) &skip_update,
    (gptr*) &skip_update, 0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
+  {"transactional", 'T', "Test in transactional mode. (Only works with block format)",
+   (gptr*) &transactional, (gptr*) &transactional, 0, GET_BOOL, NO_ARG,
+   0, 0, 0, 0, 0, 0},
   {"unique", 'C', "Undocumented", (gptr*) &opt_unique, (gptr*) &opt_unique, 0,
    GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
   {"update-rows", 'u', "Undocumented", (gptr*) &update_count,

storage/maria/ma_test2.c

@@ -46,7 +46,7 @@ static void copy_key(struct st_maria_info *info,uint inx,
 static	int verbose=0,testflag=0,
 	    first_key=0,async_io=0,pagecacheing=0,write_cacheing=0,locking=0,
             rec_pointer_size=0,pack_fields=1,silent=0,
-            opt_quick_mode=0;
+            opt_quick_mode=0, transactional= 0;
 static int pack_seg=HA_SPACE_PACK,pack_type=HA_PACK_KEY,remove_count=-1;
 static int create_flag= 0, srand_arg= 0;
 static ulong pagecache_size=IO_SIZE*16;
@@ -209,6 +209,7 @@ int main(int argc, char *argv[])
 				  (1L << (rec_pointer_size*8))/
 				  reclength : 0);
   create_info.reloc_rows=(ha_rows) 100;
+  create_info.transactional= transactional;
   if (maria_create(filename, record_type, keys,&keyinfo[first_key],
 		use_blob ? 7 : 6, &recinfo[0],
 		0,(MARIA_UNIQUEDEF*) 0,
@@ -993,6 +994,9 @@ static void get_options(int argc, char **argv)
     case 't':
       testflag=atoi(++pos);		/* testmod */
       break;
+    case 'T':
+      transactional= 1;
+      break;
     case 'q':
       opt_quick_mode=1;
       break;
@@ -1007,7 +1011,7 @@ static void get_options(int argc, char **argv)
     case 'V':
       printf("%s  Ver 1.0 for %s at %s\n",progname,SYSTEM_TYPE,MACHINE_TYPE);
       puts("By Monty, for your professional use\n");
-      printf("Usage: %s [-?AbBcDIKLPRqSsVWltv] [-k#] [-f#] [-m#] [-e#] [-E#] [-t#]\n",
+      printf("Usage: %s [-?AbBcDIKLPRqSsTVWltv] [-k#] [-f#] [-m#] [-e#] [-E#] [-t#]\n",
 	     progname);
       exit(0);
     case '#':

storage/maria/ma_test_all.sh

@@ -162,6 +162,9 @@ run_pack_tests -S
 echo "Running tests with block row format"
 run_tests -M
 
+echo "Running tests with block row format and transactions"
+run_tests "-M -T"
+
 #
 # Tests that gives warnings
 #

storage/maria/ma_update.c

@@ -162,7 +162,7 @@ int maria_update(register MARIA_HA *info, const byte *oldrec, byte *newrec)
     memcpy((char*) &state, (char*) info->state, sizeof(state));
     org_split=	     share->state.split;
     org_delete_link= share->state.dellink;
-    if ((*share->update_record)(info,pos,newrec))
+    if ((*share->update_record)(info, pos, oldrec, newrec))
       goto err;
     if (!key_changed &&
 	(memcmp((char*) &state, (char*) info->state, sizeof(state)) ||

storage/maria/maria_def.h

@@ -25,10 +25,14 @@
 #include <my_no_pthread.h>
 #endif
 
-#include <pagecache.h>
 #include "ma_loghandler.h"
 #include "ma_control_file.h"
 
+#define MAX_NONMAPPED_INSERTS 1000
+#define MARIA_MAX_TREE_LEVELS 32
+
+struct st_transaction;
+
 /* undef map from my_nosys; We need test-if-disk full */
 #undef my_write	
 
@@ -205,8 +209,6 @@ typedef struct st_maria_file_bitmap
 } MARIA_FILE_BITMAP;
 
 
-#define MAX_NONMAPPED_INSERTS 1000
-
 typedef struct st_maria_share
 {					/* Shared between opens */
   MARIA_STATE_INFO state;
@@ -250,8 +252,8 @@ typedef struct st_maria_share
   /* Called when write failed */
   my_bool (*write_record_abort)(struct st_maria_info *);
   my_bool (*update_record)(struct st_maria_info *, MARIA_RECORD_POS,
-                           const byte *);
-  my_bool (*delete_record)(struct st_maria_info *);
+                           const byte *, const byte *);
+  my_bool (*delete_record)(struct st_maria_info *, const byte *record);
   my_bool (*compare_record)(struct st_maria_info *, const byte *);
   /* calculate checksum for a row */
   ha_checksum(*calc_checksum)(struct st_maria_info *, const byte *);
@@ -288,6 +290,7 @@ typedef struct st_maria_share
   uint base_length;
   myf write_flag;
   enum data_file_type data_file_type;
+  enum pagecache_page_type page_type;   /* value depending transactional */
   my_bool temporary;
   /* Below flag is needed to make log tables work with concurrent insert */
   my_bool is_log_table;
@@ -345,7 +348,6 @@ typedef struct st_maria_row
   MARIA_RECORD_POS *tail_positions;
   ha_checksum checksum;
   byte *empty_bits, *field_lengths;
-  byte *empty_bits_buffer;              /* For storing cur_row.empty_bits */
   uint *null_field_lengths;             /* All null field lengths */
   ulong *blob_lengths;                  /* Length for each blob */
   ulong base_length, normal_length, char_length, varchar_length, blob_length;
@@ -371,6 +373,7 @@ typedef struct st_maria_block_scan
 struct st_maria_info
 {
   MARIA_SHARE *s;			/* Shared between open:s */
+  struct st_transaction *trn;           /* Pointer to active transaction */
   MARIA_STATUS_INFO *state, save_state;
   MARIA_ROW cur_row;                    /* The active row that we just read */
   MARIA_ROW new_row;			/* Storage for a row during update */
@@ -378,8 +381,10 @@ struct st_maria_info
   MARIA_BLOB *blobs;			/* Pointer to blobs */
   MARIA_BIT_BUFF bit_buff;
   DYNAMIC_ARRAY bitmap_blocks;
+  DYNAMIC_ARRAY pinned_pages;
   /* accumulate indexfile changes between write's */
   TREE *bulk_insert;
+  LEX_STRING *log_row_parts;		/* For logging */
   DYNAMIC_ARRAY *ft1_to_ft2;		/* used only in ft1->ft2 conversion */
   MEM_ROOT      ft_memroot;             /* used by the parser               */
   MYSQL_FTPARSER_PARAM *ftparser_param;	/* share info between init/deinit */
@@ -391,6 +396,7 @@ struct st_maria_info
   byte *rec_buff;			/* Temp buffer for recordpack */
   byte *int_keypos,			/* Save position for next/previous */
    *int_maxpos;				/* -""- */
+  byte *update_field_data;		/* Used by update in rows-in-block */
   uint int_nod_flag;			/* -""- */
   uint32 int_keytree_version;		/* -""- */
   int (*read_record) (struct st_maria_info *, byte*, MARIA_RECORD_POS);
@@ -568,8 +574,8 @@ extern pthread_mutex_t THR_LOCK_maria;
 #define rw_unlock(A) {}
 #endif
 
-	/* Some extern variables */
 
+/* Some extern variables */
 extern LIST *maria_open_list;
 extern uchar NEAR maria_file_magic[], NEAR maria_pack_file_magic[];
 extern uint NEAR maria_read_vec[], NEAR maria_readnext_vec[];
@@ -578,8 +584,8 @@ extern const char *maria_data_root;
 extern byte maria_zero_string[];
 extern my_bool maria_inited;
 
-	/* This is used by _ma_calc_xxx_key_length och _ma_store_key */
 
+/* This is used by _ma_calc_xxx_key_length och _ma_store_key */
 typedef struct st_maria_s_param
 {
   uint ref_length, key_length, n_ref_length;
@@ -589,26 +595,34 @@ typedef struct st_maria_s_param
   bool store_not_null;
 } MARIA_KEY_PARAM;
 
-	/* Prototypes for intern functions */
 
+/* Used to store reference to pinned page */
+typedef struct st_pinned_page
+{
+  PAGECACHE_PAGE_LINK link;
+  enum pagecache_page_lock unlock;
+} MARIA_PINNED_PAGE;
+
+
+/* Prototypes for intern functions */
 extern int _ma_read_dynamic_record(MARIA_HA *, byte *, MARIA_RECORD_POS);
 extern int _ma_read_rnd_dynamic_record(MARIA_HA *, byte *, MARIA_RECORD_POS,
                                        my_bool);
 extern my_bool _ma_write_dynamic_record(MARIA_HA *, const byte *);
 extern my_bool _ma_update_dynamic_record(MARIA_HA *, MARIA_RECORD_POS,
-                                         const byte *);
-extern my_bool _ma_delete_dynamic_record(MARIA_HA *info);
+                                         const byte *, const byte *);
+extern my_bool _ma_delete_dynamic_record(MARIA_HA *info, const byte *record);
 extern my_bool _ma_cmp_dynamic_record(MARIA_HA *info, const byte *record);
 extern my_bool _ma_write_blob_record(MARIA_HA *, const byte *);
 extern my_bool _ma_update_blob_record(MARIA_HA *, MARIA_RECORD_POS,
-                                      const byte *);
+                                      const byte *, const byte *);
 extern int _ma_read_static_record(MARIA_HA *info, byte *, MARIA_RECORD_POS);
 extern int _ma_read_rnd_static_record(MARIA_HA *, byte *, MARIA_RECORD_POS,
                                       my_bool);
 extern my_bool _ma_write_static_record(MARIA_HA *, const byte *);
 extern my_bool _ma_update_static_record(MARIA_HA *, MARIA_RECORD_POS,
-                                        const byte *);
-extern my_bool _ma_delete_static_record(MARIA_HA *info);
+                                        const byte *, const byte *);
+extern my_bool _ma_delete_static_record(MARIA_HA *info, const byte *record);
 extern my_bool _ma_cmp_static_record(MARIA_HA *info, const byte *record);
 extern int _ma_ck_write(MARIA_HA *info, uint keynr, byte *key,
                         uint length);
@@ -891,3 +905,8 @@ int _ma_create_index_by_sort(MARIA_SORT_PARAM *info, my_bool no_messages,
                              ulong);
 int _ma_sync_table_files(const MARIA_HA *info);
 int _ma_initialize_data_file(File dfile, MARIA_SHARE *share);
+
+void _ma_unpin_all_pages(MARIA_HA *info, LSN undo_lsn);
+
+extern PAGECACHE *maria_log_pagecache;
+

storage/maria/trnman.h

@@ -16,10 +16,12 @@
 #ifndef _trnman_h
 #define _trnman_h
 
-#include "lockman.h"
+C_MODE_START
 
-typedef uint64 TrID; /* our TrID is 6 bytes */
-typedef struct st_transaction TRN;
+#include <lf.h>
+#include "lockman.h"
+#include "trnman_public.h"
+#include "ma_loghandler_lsn.h"
 
 /*
   trid - 6 byte transaction identifier. Assigned when a transaction
@@ -29,28 +31,28 @@ typedef struct st_transaction TRN;
   short_trid - 2-byte transaction identifier, identifies a running
   transaction, is reassigned when transaction ends.
 */
+
+/*
+  short transaction id is at the same time its identifier
+  for a lock manager - its lock owner identifier (loid)
+*/
+
+#define short_id locks.loid
+
 struct st_transaction
 {
   LOCK_OWNER           locks; /* must be the first! see short_trid_to_TRN() */
   LF_PINS             *pins;
   TrID                 trid, min_read_from, commit_trid;
   TRN                 *next, *prev;
+  LSN		       undo_lsn;
+  uint                 locked_tables;
   /* Note! if locks.loid is 0, trn is NOT initialized */
 };
 
-#define SHORT_TRID_MAX 65535
-
-extern uint trnman_active_transactions, trnman_allocated_transactions;
+TRN dummy_transaction_object;
 
-int trnman_init(void);
-void trnman_destroy(void);
-TRN *trnman_new_trn(pthread_mutex_t *mutex, pthread_cond_t *cond);
-void trnman_end_trn(TRN *trn, my_bool commit);
-#define trnman_commit_trn(T) trnman_end_trn(T, TRUE)
-#define trnman_abort_trn(T)  trnman_end_trn(T, FALSE)
-void trnman_free_trn(TRN *trn);
-my_bool trnman_can_read_from(TRN *trn, TrID trid);
-my_bool trnman_collect_transactions(LEX_STRING *str_act, LEX_STRING *str_com);
+C_MODE_END
 
 #endif
 

unittest/mytap/tap.c

@@ -166,9 +166,15 @@ static signal_entry install_signal[]= {
 #endif
 };
 
+int skip_big_tests= 0;
+
 void
 plan(int const count)
 {
+  char *config= getenv("MYTAP_CONFIG");
+
+  if (config)
+    skip_big_tests= strcmp(config, "big");
 
   setvbuf(tapout, 0, _IONBF, 0);  /* provide output at once */
   /*