Changes needed for ColumnStore and insert cache

MCOL-3875 Columnstore write cache

The main change is to change thr_lock function get_status to
return a value that indicates we have to abort the lock.

Other thing:
- Made start_bulk_insert() and end_bulk_insert() protected so that the
  insert cache can use these

include/thr_lock.h

@@ -126,7 +126,7 @@ typedef struct st_thr_lock {
   /* write_lock_count is incremented for write locks and reset on read locks */
   ulong write_lock_count;
   uint read_no_write_count;
-  void (*get_status)(void*, my_bool);	/* When one gets a lock */
+  my_bool (*get_status)(void*, my_bool);/* Called when one gets a lock */
   void (*copy_status)(void*,void*);
   void (*update_status)(void*);		/* Before release of write */
   void (*restore_status)(void*);        /* Before release of read */

mysys/thr_lock.c

@@ -634,9 +634,10 @@ wait_for_lock(struct st_lock_list *wait, THR_LOCK_DATA *data,
   else
   {
     result= THR_LOCK_SUCCESS;
-    if (data->lock->get_status)
-      (*data->lock->get_status)(data->status_param,
-                                data->type == TL_WRITE_CONCURRENT_INSERT);
+    if (data->lock->get_status &&
+        (*data->lock->get_status)(data->status_param,
+                                  data->type == TL_WRITE_CONCURRENT_INSERT))
+        result= THR_LOCK_ABORTED;
     check_locks(data->lock,"got wait_for_lock", data->type, 0);
   }
   mysql_mutex_unlock(&data->lock->mutex);
@@ -811,8 +812,8 @@ thr_lock(THR_LOCK_DATA *data, THR_LOCK_INFO *owner, ulong lock_wait_timeout)
 	if (lock_type == TL_READ_NO_INSERT)
 	  lock->read_no_write_count++;
 	check_locks(lock,"read lock with old write lock", lock_type, 0);
-	if (lock->get_status)
-	  (*lock->get_status)(data->status_param, 0);
+	if ((lock->get_status) && (*lock->get_status)(data->status_param, 0))
+          result= THR_LOCK_ABORTED;
 	statistic_increment(locks_immediate,&THR_LOCK_lock);
 	goto end;
       }
@@ -835,8 +836,8 @@ thr_lock(THR_LOCK_DATA *data, THR_LOCK_INFO *owner, ulong lock_wait_timeout)
       if (lock_type == TL_READ_NO_INSERT)
 	lock->read_no_write_count++;
       check_locks(lock,"read lock with no write locks", lock_type, 0);
-      if (lock->get_status)
-	(*lock->get_status)(data->status_param, 0);
+      if ((lock->get_status) && (*lock->get_status)(data->status_param, 0))
+        result= THR_LOCK_ABORTED;
       statistic_increment(locks_immediate,&THR_LOCK_lock);
       goto end;
     }
@@ -951,9 +952,10 @@ thr_lock(THR_LOCK_DATA *data, THR_LOCK_INFO *owner, ulong lock_wait_timeout)
 	data->prev=lock->write.last;
 	lock->write.last= &data->next;
 	check_locks(lock,"second write lock", lock_type, 0);
-	if (lock->get_status)
-	  (*lock->get_status)(data->status_param,
-                              lock_type == TL_WRITE_CONCURRENT_INSERT);
+	if ((lock->get_status) &&
+            (*lock->get_status)(data->status_param,
+                                lock_type == TL_WRITE_CONCURRENT_INSERT))
+          result= THR_LOCK_ABORTED;
 	statistic_increment(locks_immediate,&THR_LOCK_lock);
 	goto end;
       }
@@ -986,8 +988,9 @@ thr_lock(THR_LOCK_DATA *data, THR_LOCK_INFO *owner, ulong lock_wait_timeout)
 	  (*lock->write.last)=data;		/* Add as current write lock */
 	  data->prev=lock->write.last;
 	  lock->write.last= &data->next;
-	  if (lock->get_status)
-	    (*lock->get_status)(data->status_param, concurrent_insert);
+	  if ((lock->get_status) &&
+              (*lock->get_status)(data->status_param, concurrent_insert))
+            result= THR_LOCK_ABORTED;
 	  check_locks(lock,"only write lock", lock_type, 0);
 	  statistic_increment(locks_immediate,&THR_LOCK_lock);
 	  goto end;
@@ -1581,6 +1584,7 @@ my_bool thr_upgrade_write_delay_lock(THR_LOCK_DATA *data,
   {
     if (!lock->read.data)			/* No read locks */
     {						/* We have the lock */
+      /* For this function, get_status is not allowed to fail */
       if (data->lock->get_status)
 	(*data->lock->get_status)(data->status_param, 0);
       mysql_mutex_unlock(&lock->mutex);
@@ -1781,9 +1785,10 @@ static ulong sum=0;
 
 /* The following functions is for WRITE_CONCURRENT_INSERT */
 
-static void test_get_status(void* param __attribute__((unused)),
-                            my_bool concurrent_insert __attribute__((unused)))
+static my_bool test_get_status(void* param __attribute__((unused)),
+                               my_bool concurrent_insert __attribute__((unused)))
 {
+  return 0;
 }
 
 static void test_update_status(void* param __attribute__((unused)))

sql/handler.h

@@ -4866,9 +4866,9 @@ class handler :public Sql_alloc
     DBUG_ASSERT(!(ha_table_flags() & HA_CAN_REPAIR));
     return HA_ADMIN_NOT_IMPLEMENTED;
   }
+protected:
   virtual void start_bulk_insert(ha_rows rows, uint flags) {}
   virtual int end_bulk_insert() { return 0; }
-protected:
   virtual int index_read(uchar * buf, const uchar * key, uint key_len,
                          enum ha_rkey_function find_flag)
    { return  HA_ERR_WRONG_COMMAND; }

storage/csv/ha_tina.cc

@@ -883,10 +883,11 @@ int ha_tina::find_current_row(uchar *buf)
   for CSV engine. For more details see mysys/thr_lock.c
 */
 
-void tina_get_status(void* param, my_bool concurrent_insert)
+my_bool tina_get_status(void* param, my_bool concurrent_insert)
 {
   ha_tina *tina= (ha_tina*) param;
   tina->get_status();
+  return 0;
 }
 
 void tina_update_status(void* param)

storage/maria/ha_maria.h

@@ -40,7 +40,9 @@ extern ulonglong maria_recover_options;
 
 class ha_maria :public handler
 {
+public:
   MARIA_HA *file;
+private:
   ulonglong int_table_flags;
   MARIA_RECORD_POS remember_pos;
   char *data_file_name, *index_file_name;

storage/maria/ma_blockrec.h

@@ -306,7 +306,7 @@ my_bool write_hook_for_file_id(enum translog_record_type type,
 my_bool write_hook_for_commit(enum translog_record_type type,
                               TRN *trn, MARIA_HA *tbl_info, LSN *lsn,
                               void *hook_arg);
-void _ma_block_get_status(void *param, my_bool concurrent_insert);
+my_bool _ma_block_get_status(void *param, my_bool concurrent_insert);
 my_bool _ma_block_start_trans(void* param);
 my_bool _ma_block_start_trans_no_versioning(void *param);
 void _ma_block_update_status(void *param);

storage/maria/ma_check.c

@@ -2372,7 +2372,7 @@ static int initialize_variables_for_repair(HA_CHECK *param,
 
   /* Repair code relies on share->state.state so we have to update it here */
   if (share->lock.update_status)
-    (*share->lock.update_status)(info);
+    (*share->lock.update_status)(info->lock.status_param);
 
   bzero((char*) sort_info,  sizeof(*sort_info));
   bzero((char*) sort_param, sizeof(*sort_param));

storage/maria/ma_state.c

@@ -282,7 +282,7 @@ void _ma_reset_state(MARIA_HA *info)
 			(THR_WRITE_CONCURRENT_INSERT was used)
 */
 
-void _ma_get_status(void* param, my_bool concurrent_insert)
+my_bool _ma_get_status(void* param, my_bool concurrent_insert)
 {
   MARIA_HA *info=(MARIA_HA*) param;
   DBUG_ENTER("_ma_get_status");
@@ -301,7 +301,7 @@ void _ma_get_status(void* param, my_bool concurrent_insert)
   info->state= &info->state_save;
   info->state->changed= 0;
   info->append_insert_at_end= concurrent_insert;
-  DBUG_VOID_RETURN;
+  DBUG_RETURN(0);
 }
 
 
@@ -359,7 +359,7 @@ void _ma_update_status_with_lock(MARIA_HA *info)
     locked= 1;
     mysql_mutex_lock(&info->s->lock.mutex);
   }
-  (*info->s->lock.update_status)(info);
+  (*info->s->lock.update_status)(info->lock.status_param);
   if (locked)
     mysql_mutex_unlock(&info->s->lock.mutex);
 }
@@ -379,11 +379,12 @@ void _ma_copy_status(void* to, void *from)
 }
 
 
-void _ma_reset_update_flag(void *param,
-                           my_bool concurrent_insert __attribute__((unused)))
+my_bool _ma_reset_update_flag(void *param,
+                              my_bool concurrent_insert __attribute__((unused)))
 {
   MARIA_HA *info=(MARIA_HA*) param;
   info->state->changed= 0;
+  return 0;
 }
 
 my_bool _ma_start_trans(void* param)
@@ -628,7 +629,7 @@ void _ma_remove_table_from_trnman(MARIA_HA *info)
 			(THR_WRITE_CONCURRENT_INSERT was used)
 */
 
-void _ma_block_get_status(void* param, my_bool concurrent_insert)
+my_bool _ma_block_get_status(void* param, my_bool concurrent_insert)
 {
   MARIA_HA *info=(MARIA_HA*) param;
   DBUG_ENTER("_ma_block_get_status");
@@ -647,7 +648,7 @@ void _ma_block_get_status(void* param, my_bool concurrent_insert)
   {
     DBUG_ASSERT(info->lock.type != TL_WRITE_CONCURRENT_INSERT);
   }
-  DBUG_VOID_RETURN;
+  DBUG_RETURN(0);
 }
 
 

storage/maria/ma_state.h

@@ -61,15 +61,15 @@ MARIA_STATE_HISTORY *_ma_remove_not_visible_states(MARIA_STATE_HISTORY
                                                    my_bool all,
                                                    my_bool trman_is_locked);
 void _ma_reset_state(MARIA_HA *info);
-void _ma_get_status(void* param, my_bool concurrent_insert);
+my_bool _ma_get_status(void* param, my_bool concurrent_insert);
 void _ma_update_status(void* param);
 void _ma_update_status_with_lock(MARIA_HA *info);
 void _ma_restore_status(void *param);
 void _ma_copy_status(void* to, void *from);
-void _ma_reset_update_flag(void *param, my_bool concurrent_insert);
+my_bool _ma_reset_update_flag(void *param, my_bool concurrent_insert);
 my_bool _ma_start_trans(void* param);
 my_bool _ma_check_status(void *param);
-void _ma_block_get_status(void* param, my_bool concurrent_insert);
+my_bool _ma_block_get_status(void* param, my_bool concurrent_insert);
 void _ma_block_update_status(void *param);
 void _ma_block_restore_status(void *param);
 my_bool _ma_block_check_status(void *param);

storage/maria/maria_def.h

@@ -1370,7 +1370,7 @@ my_bool _ma_cmp_dynamic_unique(MARIA_HA *info, MARIA_UNIQUEDEF *def,
                                const uchar *record, MARIA_RECORD_POS pos);
 my_bool _ma_unique_comp(MARIA_UNIQUEDEF *def, const uchar *a, const uchar *b,
                         my_bool null_are_equal);
-void _ma_get_status(void *param, my_bool concurrent_insert);
+my_bool _ma_get_status(void *param, my_bool concurrent_insert);
 void _ma_update_status(void *param);
 void _ma_restore_status(void *param);
 void _ma_copy_status(void *to, void *from);

storage/myisam/mi_locking.c

@@ -284,7 +284,7 @@ int mi_lock_database(MI_INFO *info, int lock_type)
 			(THR_WRITE_CONCURRENT_INSERT was used)
 */
 
-void mi_get_status(void* param, my_bool concurrent_insert)
+my_bool mi_get_status(void* param, my_bool concurrent_insert)
 {
   MI_INFO *info=(MI_INFO*) param;
   DBUG_ENTER("mi_get_status");
@@ -306,7 +306,7 @@ void mi_get_status(void* param, my_bool concurrent_insert)
   info->append_insert_at_end= concurrent_insert;
   if (concurrent_insert)
     info->s->state.state.uncacheable= TRUE;
-  DBUG_VOID_RETURN;
+  DBUG_RETURN(0);
 }
 
 

storage/myisam/myisamdef.h

@@ -726,7 +726,7 @@ int _mi_cmp_dynamic_unique(MI_INFO *info, MI_UNIQUEDEF *def,
                            const uchar *record, my_off_t pos);
 int mi_unique_comp(MI_UNIQUEDEF *def, const uchar *a, const uchar *b,
                    my_bool null_are_equal);
-void mi_get_status(void *param, my_bool concurrent_insert);
+my_bool mi_get_status(void *param, my_bool concurrent_insert);
 void mi_update_status(void *param);
 void mi_restore_status(void *param);
 void mi_copy_status(void *to, void *from);