Fix for Bug#40311 Assert in MARIA_RECORD_POS during pushbuild 2 test:

Fixed bug when removing a newly inserted record (in case of duplicate key).
The bug caused a crash for rows with several blobs and the first blob was small enough to fit into the head page.
Don't change state_history if nothing changed (speed optimization that also simplifies logic).
Reset state_history if we added/deleted or updated rows without versioning.
Fixed wrong test in trnman_exists_active_transactions() if state is visible or not.

Other bugs fixed:
Fixed wrong argument to (lock->get_status) when we had to wait for TL_WRITE_CONCURRENT_INSERT.
Item_equal::update_used_tables() didn't calculate const_item_cache properly.
Added assert's to detect if join_read_const_table() was called under wrong assumptions..
Fixed that _ma_setup_live_state() is called from thr_lock() instead of handler::external_lock().
This was needed to get versioning information to be setup correctly.
Fixed error in debug binaries during a call to _ma_check_table_is_closed() when another thread was opening/closing a table.
Fixed wrong test when finding right history_state to use.

mysql-test/suite/maria/r/maria.result:
  Added test for Bug#40311 Assert in MARIA_RECORD_POS during pushbuild 2 test
mysql-test/suite/maria/t/maria.test:
  Added test for Bug#40311 Assert in MARIA_RECORD_POS during pushbuild 2 test
mysys/thr_lock.c:
  Fixed wrong argument to (lock->get_status) when we had to wait for TL_WRITE_CONCURRENT_INSERT
sql/item_cmpfunc.cc:
  Item_equal::update_used_tables() didn't calculate const_item_cache properly, which later caused a wrong result for item->const_item()
sql/sql_base.cc:
  In debug mode, Initilize record buffer with unexpected data to catch usage of uninitialized memory
sql/sql_select.cc:
  Fixed indentation
  Added assert's to detect if join_read_const_table() was called under wrong assumptions.
  One assert() is disabled for now as Item_equal() doesn't behave as expected.
storage/maria/ha_maria.cc:
  Move calling to _ma_setup_live_state() to ma_state.c::_ma_block_get_status()
  This was needed as _ma_setup_live_state() needed to know if the table will be used concurrently or not
storage/maria/ma_blockrec.c:
  Fixed bug when removing a newly inserted record (in case of duplicate key).
  The bug caused a crash for rows with several blobs and the first blob was small enough to fit into the head page.
storage/maria/ma_dbug.c:
  Added mutex to protect the open table list during _ma_check_table_is_closed().
  Without the protection we could get a error in debug binaries during a call to _ma_check_table_is_closed()
storage/maria/ma_delete_table.c:
  Removed not used code
storage/maria/ma_rename.c:
  Removed not used code
storage/maria/ma_state.c:
  Fixed wrong test when finding right history_state to use
  Mark in tables->state_current.no_transid if we are using transid's or not.
  Don't change state_history if nothing changed (speed optimization that also simplifies logic)
  Reset state_history if we added/deleted or updated rows without versioning.
  More DBUG_ASSERT's and more DBUG
  Updated maria_versioning() to initialize environment before calling _ma_blok_get_status(). This was needed because of the new logic in _ma_block_get_status()
storage/maria/ma_state.h:
  Added flags to detect if table changed and/or if we changed table without versioning
storage/maria/ma_write.c:
  Simple cleanups (No logic changes)
storage/maria/trnman.c:
  Fixed wrong test in trnman_exists_active_transactions() if state is visible or not.

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

@@ -2583,3 +2583,10 @@ lock table t1 read, t2 read;
 flush tables with read lock;
 unlock tables;
 drop table t1, t2;
+create table t1(a int primary key, b blob, c blob) engine=maria;
+insert into t1 values(1,repeat('a',100), repeat('b',657860));
+Warnings:
+Warning	1265	Data truncated for column 'c' at row 1
+insert into t1 values(1,repeat('a',100), repeat('b',657860));
+ERROR 23000: Duplicate entry '1' for key 'PRIMARY'
+drop table t1;

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

@@ -1846,6 +1846,17 @@ flush tables with read lock;
 unlock tables;
 drop table t1, t2;
 
+#
+# Bug #40311
+# Crash when aborting inserting of row with 2 blobs where first is short
+#
+
+create table t1(a int primary key, b blob, c blob) engine=maria;
+insert into t1 values(1,repeat('a',100), repeat('b',657860));
+--error ER_DUP_ENTRY
+insert into t1 values(1,repeat('a',100), repeat('b',657860));
+drop table t1;
+
 # Set defaults back
 --disable_result_log
 --disable_query_log

mysys/thr_lock.c

@@ -494,7 +494,8 @@ wait_for_lock(struct st_lock_list *wait, THR_LOCK_DATA *data,
   {
     result= THR_LOCK_SUCCESS;
     if (data->lock->get_status)
-      (*data->lock->get_status)(data->status_param, 0);
+      (*data->lock->get_status)(data->status_param,
+                                data->type == TL_WRITE_CONCURRENT_INSERT);
     check_locks(data->lock,"got wait_for_lock",0);
   }
   pthread_mutex_unlock(&data->lock->mutex);

sql/item_cmpfunc.cc

@@ -5234,6 +5234,7 @@ void Item_equal::update_used_tables()
   not_null_tables_cache= used_tables_cache= 0;
   if ((const_item_cache= cond_false))
     return;
+  const_item_cache= 1;
   while ((item=li++))
   {
     item->update_used_tables();

sql/sql_base.cc

@@ -2979,6 +2979,13 @@ TABLE *open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
   table->pos_in_table_list= table_list;
   table_list->updatable= 1; // It is not derived table nor non-updatable VIEW
   table->clear_column_bitmaps();
+#if !defined(DBUG_OFF) && !defined(HAVE_purify)
+  /*
+    Fill record with random values to find bugs where we access fields
+    without first reading them.
+  */
+  bfill(table->record[0], table->reclength, 254);
+#endif
   DBUG_ASSERT(table->key_read == 0);
   DBUG_RETURN(table);
 }

sql/sql_select.cc

@@ -2517,11 +2517,10 @@ make_join_statistics(JOIN *join, TABLE_LIST *tables, COND *conds,
     s->needed_reg.init();
     table_vector[i]=s->table=table=tables->table;
     table->pos_in_table_list= tables;
-    error= table->file->info(HA_STATUS_VARIABLE | HA_STATUS_NO_LOCK);
-    if(error)
+    if ((error= table->file->info(HA_STATUS_VARIABLE | HA_STATUS_NO_LOCK)))
     {
-        table->file->print_error(error, MYF(0));
-        DBUG_RETURN(1);
+      table->file->print_error(error, MYF(0));
+      DBUG_RETURN(1);
     }
     table->quick_keys.clear_all();
     table->reginfo.join_tab=s;
@@ -11609,6 +11608,11 @@ join_read_const_table(JOIN_TAB *tab, POSITION *pos)
       if (!table->maybe_null || error > 0)
 	DBUG_RETURN(error);
     }
+    /*
+      The optimizer trust the engine that when stats.records is 0, there
+      was no found rows
+    */
+    DBUG_ASSERT(table->file->stats.records > 0 || error);
   }
   else
   {
@@ -11638,6 +11642,17 @@ join_read_const_table(JOIN_TAB *tab, POSITION *pos)
   }
   if (*tab->on_expr_ref && !table->null_row)
   {
+#if !defined(DBUG_OFF) && defined(NOT_USING_ITEM_EQUAL)
+    /*
+      This test could be very usefull to find bugs in the optimizer
+      where we would call this function with an expression that can't be
+      evaluated yet. We can't have this enabled by default as long as
+      have items like Item_equal, that doesn't report they are const but
+      they can still be called even if they contain not const items.
+    */
+    (*tab->on_expr_ref)->update_used_tables();
+    DBUG_ASSERT((*tab->on_expr_ref)->const_item());
+#endif
     if ((table->null_row= test((*tab->on_expr_ref)->val_int() == 0)))
       mark_as_null_row(table);  
   }

storage/maria/ha_maria.cc

@@ -2319,13 +2319,7 @@ int ha_maria::external_lock(THD *thd, int lock_type)
         trnman_new_statement(trn);
       }
 
-      /* If handler uses versioning */
-      if (file->s->lock_key_trees)
-      {
-        if (_ma_setup_live_state(file))
-          DBUG_RETURN(HA_ERR_OUT_OF_MEM);
-      }
-      else
+      if (!file->s->lock_key_trees)             // If we don't use versioning
       {
         /*
           We come here in the following cases:

storage/maria/ma_blockrec.c

@@ -3489,23 +3489,26 @@ my_bool _ma_write_abort_block_record(MARIA_HA *info)
   for (block= blocks->block + 1, end= block + blocks->count - 1; block < end;
        block++)
   {
-    if (block->used & BLOCKUSED_TAIL)
+    if (block->used & BLOCKUSED_USED)
     {
-      /*
-        block->page_count is set to the tail directory entry number in
-        write_block_record()
-      */
-      if (delete_head_or_tail(info, block->page, block->page_count & ~TAIL_BIT,
-                              0, 0))
-        res= 1;
-    }
-    else if (block->used & BLOCKUSED_USED)
-    {
-      if (free_full_page_range(info, block->page, block->page_count))
-        res= 1;
+      if (block->used & BLOCKUSED_TAIL)
+      {
+        /*
+          block->page_count is set to the tail directory entry number in
+          write_block_record()
+        */
+        if (delete_head_or_tail(info, block->page,
+                                block->page_count & ~TAIL_BIT,
+                                0, 0))
+          res= 1;
+      }
+      else
+      {
+        if (free_full_page_range(info, block->page, block->page_count))
+          res= 1;
+      }
     }
   }
-
   if (share->now_transactional)
   {
     if (_ma_write_clr(info, info->cur_row.orig_undo_lsn,

storage/maria/ma_dbug.c

@@ -180,6 +180,7 @@ my_bool _ma_check_table_is_closed(const char *name, const char *where)
   DBUG_ENTER("_ma_check_table_is_closed");
 
   (void) fn_format(filename,name,"",MARIA_NAME_IEXT,4+16+32);
+  pthread_mutex_lock(&THR_LOCK_maria);
   for (pos=maria_open_list ; pos ; pos=pos->next)
   {
     MARIA_HA *info=(MARIA_HA*) pos->data;
@@ -190,10 +191,12 @@ my_bool _ma_check_table_is_closed(const char *name, const char *where)
       {
 	fprintf(stderr,"Warning:  Table: %s is open on %s\n", name,where);
 	DBUG_PRINT("warning",("Table: %s is open on %s", name,where));
+        pthread_mutex_unlock(&THR_LOCK_maria);
 	DBUG_RETURN(1);
       }
     }
   }
+  pthread_mutex_unlock(&THR_LOCK_maria);
   DBUG_RETURN(0);
 }
 #endif /* EXTRA_DEBUG */

storage/maria/ma_delete_table.c

@@ -69,11 +69,6 @@ int maria_delete_table(const char *name)
       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)
   {

storage/maria/ma_rename.c

@@ -104,11 +104,6 @@ int maria_rename(const char *old_name, const char *new_name)
   }
 
   maria_close(info);
-#ifdef USE_RAID
-#ifdef EXTRA_DEBUG
-  _ma_check_table_is_closed(old_name,"rename raidcheck");
-#endif
-#endif /* USE_RAID */
 
   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);

storage/maria/ma_state.c

@@ -91,18 +91,26 @@ my_bool _ma_setup_live_state(MARIA_HA *info)
     It's enough to compare trids here (instead of calling
     tranman_can_read_from) as history->trid is a commit_trid
   */
-  while (trn->trid < history->trid && history->trid != ~(TrID)0)
+  while (trn->trid <= history->trid)
     history= history->next;
   pthread_mutex_unlock(&share->intern_lock);
   /* The current item can't be deleted as it's the first one visible for us */
   tables->state_start=  tables->state_current= history->state;
-  tables->state_current.changed= 0;
+  tables->state_current.changed= tables->state_current.no_transid= 0;
 
   DBUG_PRINT("info", ("records: %ld", (ulong) tables->state_start.records));
 
 end:
   info->state_start= &tables->state_start;
   info->state= &tables->state_current;
+
+  /*
+    Mark in transaction state if we are not using transid (versioning)
+    on rows. If not, then we will in _ma_trnman_end_trans_hook()
+    ensure that the state is visible for all at end of transaction
+  */
+  tables->state_current.no_transid|= !(info->row_flag & ROW_FLAG_TRANSID);
+
   DBUG_RETURN(0);
 }
 
@@ -406,51 +414,70 @@ my_bool _ma_trnman_end_trans_hook(TRN *trn, my_bool commit,
       MARIA_STATE_HISTORY *history;
 
       pthread_mutex_lock(&share->intern_lock);
-      if (active_transactions && share->now_transactional &&
-          trnman_exists_active_transactions(share->state_history->trid,
-                                            trn->commit_trid, 1))
+
+      /* We only have to update history state if something changed */
+      if (tables->state_current.changed)
       {
-        /*
-          There exist transactions that are still using the current
-          share->state_history.  Create a new history item for this
-          commit and add it first in the state_history list. This
-          ensures that all history items are stored in the list in
-          decresing trid order.
-        */
-        if (!(history= my_malloc(sizeof(*history), MYF(MY_WME))))
+        if (tables->state_current.no_transid)
         {
-          /* purecov: begin inspected */
-          error= 1;
-          pthread_mutex_unlock(&share->intern_lock);
-          my_free(tables, MYF(0));
-          continue;
-          /* purecov: end */
+          /*
+            The change was done without using transid on rows (like in
+            bulk insert). In this case this thread is the only one
+            that is using the table and all rows will be visble
+            for all transactions.
+          */
+          _ma_reset_history(share);
+        }
+        else
+        {
+          if (active_transactions && share->now_transactional &&
+              trnman_exists_active_transactions(share->state_history->trid,
+                                                trn->commit_trid, 1))
+          {
+            /*
+              There exist transactions that are still using the current
+              share->state_history.  Create a new history item for this
+              commit and add it first in the state_history list. This
+              ensures that all history items are stored in the list in
+              decresing trid order.
+            */
+            if (!(history= my_malloc(sizeof(*history), MYF(MY_WME))))
+            {
+              /* purecov: begin inspected */
+              error= 1;
+              pthread_mutex_unlock(&share->intern_lock);
+              my_free(tables, MYF(0));
+              continue;
+              /* purecov: end */
+            }
+            history->state= share->state_history->state;
+            history->next= share->state_history;
+            share->state_history= history;
+          }
+          else
+          {
+            /* Previous history can't be seen by anyone, reuse old memory */
+            history= share->state_history;
+            DBUG_PRINT("info", ("removing history->trid: %lu  new: %lu",
+                                (ulong) history->trid,
+                                (ulong) trn->commit_trid));
+          }
+
+          history->state.records+= (tables->state_current.records -
+                                    tables->state_start.records);
+          history->state.checksum+= (tables->state_current.checksum -
+                                     tables->state_start.checksum);
+          history->trid= trn->commit_trid;
+
+          if (history->next)
+          {
+            /* Remove not visible states */
+            share->state_history= _ma_remove_not_visible_states(history, 0, 1);
+          }
+          DBUG_PRINT("info", ("share: 0x%lx  in_trans: %d",
+                              (ulong) share, share->in_trans));
         }
-        history->state= share->state_history->state;
-        history->next= share->state_history;
-        share->state_history= history;
-      }
-      else
-      {
-        /* Previous history can't be seen by anyone, reuse old memory */
-        history= share->state_history;
-        DBUG_PRINT("info", ("removing history->trid: %lu  new: %lu",
-                            (ulong) history->trid, (ulong) trn->commit_trid));
-      }
-
-      history->state.records+= (tables->state_current.records -
-                                tables->state_start.records);
-      history->state.checksum+= (tables->state_current.checksum -
-                                 tables->state_start.checksum);
-      history->trid= trn->commit_trid;
-
-      if (history->next)
-      {
-        /* Remove not visible states */
-        share->state_history= _ma_remove_not_visible_states(history, 0, 1);
       }
-      DBUG_PRINT("info", ("share: 0x%lx  in_trans: %d",
-                          (ulong) share, share->in_trans));
       share->in_trans--;
       pthread_mutex_unlock(&share->intern_lock);
     }
@@ -511,7 +538,6 @@ void _ma_remove_table_from_trnman(MARIA_SHARE *share, TRN *trn)
 
 
 
-
 /****************************************************************************
   The following functions are called by thr_lock() in threaded applications
   for transactional tables.
@@ -536,9 +562,24 @@ void _ma_block_get_status(void* param, my_bool concurrent_insert)
   info->row_flag= info->s->base.default_row_flag;
   if (concurrent_insert)
   {
+    DBUG_ASSERT(info->lock.type == TL_WRITE_CONCURRENT_INSERT);
     info->row_flag|= ROW_FLAG_TRANSID;
     info->row_base_length+= TRANSID_SIZE;
   }
+  else
+  {
+    DBUG_ASSERT(info->lock.type != TL_WRITE_CONCURRENT_INSERT);
+  }
+
+  if (info->s->lock_key_trees)
+  {
+    /*
+      Assume for now that this doesn't fail (It can only fail in
+      out of memory conditions)
+      TODO: Fix this by having one extra state pre-allocated
+    */
+    (void) _ma_setup_live_state(info);
+  }
   DBUG_VOID_RETURN;
 }
 
@@ -574,7 +615,12 @@ void maria_versioning(MARIA_HA *info, my_bool versioning)
 {
   /* For now, this is a hack */
   if (info->s->have_versioning)
+  {
+    /* Assume is a non threaded application (for now) */
+    info->s->lock_key_trees= 0;
+    info->lock.type= versioning ? TL_WRITE_CONCURRENT_INSERT : TL_WRITE;
     _ma_block_get_status((void*) info, versioning);
+  }
 }
 
 
@@ -609,6 +655,7 @@ void _ma_copy_nontrans_state_information(MARIA_HA *info)
 void _ma_reset_history(MARIA_SHARE *share)
 {
   MARIA_STATE_HISTORY *history, *next;
+  DBUG_ENTER("_ma_reset_history");
 
   share->state_history->trid= 0;          /* Visibly by all */
   share->state_history->state= share->state.state;
@@ -620,6 +667,7 @@ void _ma_reset_history(MARIA_SHARE *share)
     next= history->next;
     my_free(history, MYF(0));
   }
+  DBUG_VOID_RETURN;
 }
 
 

storage/maria/ma_state.h

@@ -17,14 +17,15 @@
 
 typedef struct st_maria_status_info
 {
-  ha_rows records;				/* Rows in table */
-  ha_rows del;					/* Removed rows */
-  my_off_t empty;				/* lost space in datafile */
-  my_off_t key_empty;				/* lost space in indexfile */
+  ha_rows records;                      /* Rows in table */
+  ha_rows del;                          /* Removed rows */
+  my_off_t empty;                       /* lost space in datafile */
+  my_off_t key_empty;                   /* lost space in indexfile */
   my_off_t key_file_length;
   my_off_t data_file_length;
   ha_checksum checksum;
-  my_bool     changed;
+  uint32 changed:1,                     /* Set if table was changed */
+         no_transid:1;                  /* Set if no transid was set on rows */
 } MARIA_STATUS_INFO;
 
 

storage/maria/ma_write.c

@@ -230,9 +230,11 @@ int maria_write(MARIA_HA *info, uchar *record)
             /* running. now we wait */
             WT_RESOURCE_ID rc;
             int res;
+            const char *old_proc_info; 
 
             rc.type= &ma_rc_dup_unique;
-            rc.value= (intptr)blocker; /* TODO savepoint id when we'll have them */
+            /* TODO savepoint id when we'll have them */
+            rc.value= (intptr)blocker;
             res= wt_thd_will_wait_for(info->trn->wt, blocker->wt, & rc);
             if (res != WT_OK)
             {
@@ -240,14 +242,12 @@ int maria_write(MARIA_HA *info, uchar *record)
               my_errno= HA_ERR_LOCK_DEADLOCK;
               goto err;
             }
-            {
-              const char *old_proc_info= proc_info_hook(0,
-                    "waiting for a resource", __func__, __FILE__, __LINE__);
-
-              res= wt_thd_cond_timedwait(info->trn->wt, & blocker->state_lock);
+            old_proc_info= proc_info_hook(0,
+                                          "waiting for a resource",
+                                          __func__, __FILE__, __LINE__);
+            res= wt_thd_cond_timedwait(info->trn->wt, & blocker->state_lock);
+            proc_info_hook(0, old_proc_info, __func__, __FILE__, __LINE__);
 
-              proc_info_hook(0, old_proc_info, __func__, __FILE__, __LINE__);
-            }
             pthread_mutex_unlock(& blocker->state_lock);
             if (res != WT_OK)
             {

storage/maria/trnman.c

@@ -885,11 +885,22 @@ my_bool trnman_exists_active_transactions(TrID min_id, TrID max_id,
   for (trn= active_list_min.next; trn != &active_list_max; trn= trn->next)
   {
     /*
-      We use >= for min_id as min_id is a commit_trid and trn->trid
-      is transaction id.  In the case they are the same, then the
-      trn started after the min_id was committed.
+      We use <= for max_id as max_id is a commit_trid and trn->trid
+      is transaction id.  When calculating commit_trid we use the
+      current value of global_trid_generator.  global_trid_generator is
+      incremented for each new transaction.
+
+      For example, assuming we have
+      min_id = 5
+      max_id = 10
+
+      A trid of value 5 can't see the history event between 5 & 10
+      at it vas started before min_id 5 was committed.
+      A trid of value 10 can't see the next history event (max_id = 10)
+      as it started before this was committed. In this case it must use
+      the this event.
     */
-    if (trn->trid >= min_id && trn->trid < max_id)
+    if (trn->trid > min_id && trn->trid <= max_id)
     {
       ret= 1;
       break;