Fix for #Bug35048 "maria table corruption reported when transactional=0

Problem was that page in bitmap was marked as full even if there was free places in page directory


mysql-test/r/maria.result:
  Test case for problem with head/tail page with 255 entries
  (Bug 35048 "maria table corruption reported when transactional=0)
mysql-test/t/maria.test:
  Test case for problem with head/tail page with 255 entries
  (Bug 35048 "maria table corruption reported when transactional=0)
storage/maria/ma_blockrec.c:
  Fix to ensure that bitmap is marked 'full' when the head/tail page directory is full
storage/maria/ma_check.c:
  Better check when directory for head/tail pages are marked full (The page directory can't hold a row tail + blob tails)

mysql-test/r/maria.result

@@ -2254,3 +2254,21 @@ t1	CREATE TABLE `t1` (
   `c` char(1) DEFAULT NULL
 ) ENGINE=MARIA DEFAULT CHARSET=latin1 PAGE_CHECKSUM=1 TRANSACTIONAL=1
 drop table t1;
+create table t1 (i int auto_increment not null primary key) transactional=0;
+check table t1 extended;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+delete from t1 where i = 10;
+check table t1 extended;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+drop table t1;
+create table t1 (i int auto_increment not null primary key);
+check table t1 extended;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+delete from t1 where i = 10;
+check table t1 extended;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+drop table t1;

mysql-test/t/maria.test

@@ -1455,6 +1455,41 @@ alter table t1 engine=maria;
 show create table t1;
 drop table t1;
 
+#
+# Test problems with small rows and row_type=page 
+# Bug 35048 "maria table corruption reported when transactional=0"
+#
+
+create table t1 (i int auto_increment not null primary key) transactional=0;
+
+let $i=510;
+--disable_query_log
+while ($i)
+{
+  dec $i;
+  insert into t1 values (null);
+}
+--enable_query_log
+check table t1 extended;
+delete from t1 where i = 10;
+check table t1 extended;
+drop table t1;
+
+create table t1 (i int auto_increment not null primary key);
+
+let $i=510;
+--disable_query_log
+while ($i)
+{
+  dec $i;
+  insert into t1 values (null);
+}
+--enable_query_log
+check table t1 extended;
+delete from t1 where i = 10;
+check table t1 extended;
+drop table t1;
+
 # End of 5.1 tests
 
 --disable_result_log

storage/maria/ma_blockrec.c

@@ -691,6 +691,34 @@ static void check_directory(uchar *buff, uint block_size)
 #endif /* DBUG_OFF */
 
 
+/**
+   @brief Calculate if there is enough entries on the page
+*/
+
+my_bool enough_free_entries(uchar *buff, uint block_size, uint wanted_entries)
+{
+  uint entries= (uint) buff[DIR_COUNT_OFFSET];
+  uint needed_free_entries, free_entry;
+
+  if (entries + wanted_entries <= MAX_ROWS_PER_PAGE)
+    return 1;
+
+  /* Check if enough free entries in free list */
+  needed_free_entries= entries + wanted_entries - MAX_ROWS_PER_PAGE;
+
+  free_entry= (uint) buff[DIR_FREE_OFFSET];
+  while (free_entry != END_OF_DIR_FREE_LIST)
+  {
+    uchar *dir;
+    if (!--needed_free_entries)
+      return 1;
+    dir= dir_entry_pos(buff, block_size, free_entry);
+    free_entry= dir[3];
+  }
+  return 0;                                     /* Not enough entries */
+}
+
+
 /**
    @brief Extend a record area to fit a given size block
 
@@ -1029,6 +1057,7 @@ static uchar *find_free_position(uchar *buff, uint block_size, uint *res_rownr,
     DBUG_RETURN(dir);
   }
   /* No free places in dir; create a new one */
+
   /* Check if there is place for the directory entry */
   if (max_entry == MAX_ROWS_PER_PAGE)
     DBUG_RETURN(0);
@@ -1801,8 +1830,8 @@ static my_bool write_tail(MARIA_HA *info,
     during _ma_bitmap_find_place() allocate more entries on the tail page
     than it can hold
   */
-  block->empty_space= ((uint) (row_pos.buff)[DIR_COUNT_OFFSET] <=
-                       MAX_ROWS_PER_PAGE - 1 - share->base.blobs ?
+  block->empty_space= (enough_free_entries(row_pos.buff, share->block_size,
+                                           1 + share->base.blobs) ?
                        empty_space : 0);
   block->used= BLOCKUSED_USED | BLOCKUSED_TAIL;
 
@@ -2587,7 +2616,8 @@ static my_bool write_block_record(MARIA_HA *info,
   int2store(page_buff + EMPTY_SPACE_OFFSET, row_pos->empty_space);
   /* Mark in bitmaps how the current page was actually used */
   head_block->empty_space= row_pos->empty_space;
-  if (page_buff[DIR_COUNT_OFFSET] == MAX_ROWS_PER_PAGE)
+  if (page_buff[DIR_COUNT_OFFSET] == MAX_ROWS_PER_PAGE &&
+      page_buff[DIR_FREE_OFFSET] == END_OF_DIR_FREE_LIST)
     head_block->empty_space= 0;               /* Page is full */
   head_block->used|= BLOCKUSED_USED;
 
@@ -3881,6 +3911,15 @@ static my_bool delete_head_or_tail(MARIA_HA *info,
               info->pinned_pages.elements-1);
 
   DBUG_PRINT("info", ("empty_space: %u", empty_space));
+
+  /*
+    If there is not enough space for all possible tails, mark the
+    page full
+  */
+  if (!head && !enough_free_entries(buff, share->block_size,
+                                    1 + share->base.blobs))
+    empty_space= 0;
+
   DBUG_RETURN(_ma_bitmap_set(info, page, head, empty_space));
 }
 

storage/maria/ma_check.c

@@ -1442,7 +1442,7 @@ static int check_compressed_record(HA_CHECK *param, MARIA_HA *info, int extend,
 static int check_page_layout(HA_CHECK *param, MARIA_HA *info,
                              my_off_t page_pos, uchar *page,
                              uint row_count, uint head_empty,
-                             uint *real_rows_found)
+                             uint *real_rows_found, uint *free_slots_found)
 {
   uint empty, last_row_end, row, first_dir_entry, free_entry, block_size;
   uint free_entries, prev_free_entry;
@@ -1495,6 +1495,7 @@ static int check_page_layout(HA_CHECK *param, MARIA_HA *info,
     free_entry= dir[3];
     free_entries++;
   }
+  *free_slots_found= free_entries;
 
   /* Check directry */
   dir_entry= page+ block_size - PAGE_SUFFIX_SIZE;
@@ -1694,7 +1695,7 @@ static int check_block_record(HA_CHECK *param, MARIA_HA *info, int extend,
   uint block_size= share->block_size;
   ha_rows full_page_count, tail_count;
   my_bool full_dir;
-  uint offset_page, offset;
+  uint offset_page, offset, free_count;
 
   LINT_INIT(full_dir);
 
@@ -1791,7 +1792,11 @@ static int check_block_record(HA_CHECK *param, MARIA_HA *info, int extend,
                           row_count * DIR_ENTRY_SIZE);
       if (empty_space < share->bitmap.sizes[3])
         param->lost+= empty_space;
-      full_dir= row_count == MAX_ROWS_PER_PAGE;
+      if (check_page_layout(param, info, pos, page_buff, row_count,
+                            empty_space, &real_row_count, &free_count))
+        goto err;
+      full_dir= (row_count == MAX_ROWS_PER_PAGE &&
+                 page_buff[DIR_FREE_OFFSET] == END_OF_DIR_FREE_LIST);
       break;
     case TAIL_PAGE:
       row_count= ((uchar*) page_buff)[DIR_COUNT_OFFSET];
@@ -1799,9 +1804,13 @@ static int check_block_record(HA_CHECK *param, MARIA_HA *info, int extend,
       param->used+= block_size - empty_space;
       param->link_used+= (PAGE_HEADER_SIZE + PAGE_SUFFIX_SIZE +
                           row_count * DIR_ENTRY_SIZE);
-      full_dir= row_count == MAX_ROWS_PER_PAGE;
       if (empty_space < share->bitmap.sizes[6])
         param->lost+= empty_space;
+      if (check_page_layout(param, info, pos, page_buff, row_count,
+                            empty_space, &real_row_count, &free_count))
+        goto err;
+      full_dir= (row_count - free_count >= MAX_ROWS_PER_PAGE -
+                 share->base.blobs);
       break;
     case BLOB_PAGE:
       full_page_count++;
@@ -1830,9 +1839,6 @@ static int check_block_record(HA_CHECK *param, MARIA_HA *info, int extend,
     if ((enum en_page_type) page_type == BLOB_PAGE)
       continue;
     param->empty+= empty_space;
-    if (check_page_layout(param, info, pos, page_buff, row_count,
-                          empty_space, &real_row_count))
-      goto err;
     if ((enum en_page_type) page_type == TAIL_PAGE)
     {
       tail_count+= real_row_count;