1 files changed, 90 insertions, 20 deletions

diff --git a/sql/opt_range.cc b/sql/opt_range.cc
index d3c2ec5a132..7a3a0c94a18 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -117,6 +117,7 @@
 #include "records.h"          // init_read_record, end_read_record
 #include <m_ctype.h>
 #include "sql_select.h"
+#include "filesort.h"         // filesort_free_buffers
 
 #ifndef EXTRA_DEBUG
 #define test_rb_tree(A,B) {}
@@ -1790,8 +1791,6 @@ QUICK_RANGE_SELECT::QUICK_RANGE_SELECT(THD *thd, TABLE *table, uint key_nr,
     bzero((char*) &alloc,sizeof(alloc));
   file= head->file;
   record= head->record[0];
-  save_read_set= head->read_set;
-  save_write_set= head->write_set;
 
   /* Allocate a bitmap for used columns (Q: why not on MEM_ROOT?) */
   if (!(bitmap= (my_bitmap_map*) my_malloc(head->s->column_bitmap_size,
@@ -1861,7 +1860,6 @@ QUICK_RANGE_SELECT::~QUICK_RANGE_SELECT()
     free_root(&alloc,MYF(0));
     my_free(column_bitmap.bitmap);
   }
-  head->column_bitmaps_set(save_read_set, save_write_set);
   my_free(mrr_buf_desc);
   DBUG_VOID_RETURN;
 }
@@ -1895,7 +1893,8 @@ int QUICK_INDEX_SORT_SELECT::init()
 int QUICK_INDEX_SORT_SELECT::reset()
 {
   DBUG_ENTER("QUICK_INDEX_SORT_SELECT::reset");
-  DBUG_RETURN(read_keys_and_merge());
+  const int retval= read_keys_and_merge();
+  DBUG_RETURN(retval);
 }
 
 bool
@@ -2006,6 +2005,8 @@ int QUICK_RANGE_SELECT::init_ror_merged_scan(bool reuse_handler)
   handler *save_file= file, *org_file;
   my_bool org_key_read;
   THD *thd= head->in_use;
+  MY_BITMAP * const save_read_set= head->read_set;
+  MY_BITMAP * const save_write_set= head->write_set;
   DBUG_ENTER("QUICK_RANGE_SELECT::init_ror_merged_scan");
 
   in_ror_merged_scan= 1;
@@ -2056,6 +2057,7 @@ int QUICK_RANGE_SELECT::init_ror_merged_scan(bool reuse_handler)
   last_rowid= file->ref;
 
 end:
+  DBUG_ASSERT(head->read_set == &column_bitmap);
   /*
     We are only going to read key fields and call position() on 'file'
     The following sets head->tmp_set to only use this key and then updates
@@ -2069,7 +2071,8 @@ end:
   if (!head->no_keyread)
   {
     doing_key_read= 1;
-    head->mark_columns_used_by_index(index);
+    head->mark_columns_used_by_index_no_reset(index, head->read_set);
+    head->enable_keyread();
   }
 
   head->prepare_for_position();
@@ -2091,8 +2094,9 @@ end:
 
   head->file= org_file;
   head->key_read= org_key_read;
-  bitmap_copy(&column_bitmap, head->read_set);
-  head->column_bitmaps_set(&column_bitmap, &column_bitmap);
+
+  /* Restore head->read_set (and write_set) to what they had before the call */
+  head->column_bitmaps_set(save_read_set, save_write_set);
  
   if (reset())
   {
@@ -2142,16 +2146,27 @@ int QUICK_ROR_INTERSECT_SELECT::init_ror_merged_scan(bool reuse_handler)
       There is no use of this->file. Use it for the first of merged range
       selects.
     */
-    if (quick->init_ror_merged_scan(TRUE))
-      DBUG_RETURN(1);
+    int error= quick->init_ror_merged_scan(TRUE);
+    if (error)
+      DBUG_RETURN(error);
     quick->file->extra(HA_EXTRA_KEYREAD_PRESERVE_FIELDS);
   }
   while ((cur= quick_it++))
   {
     quick= cur->quick;
+#ifndef DBUG_OFF
+    const MY_BITMAP * const save_read_set= quick->head->read_set;
+    const MY_BITMAP * const save_write_set= quick->head->write_set;
+#endif
     if (quick->init_ror_merged_scan(FALSE))
       DBUG_RETURN(1);
     quick->file->extra(HA_EXTRA_KEYREAD_PRESERVE_FIELDS);
+
+    // Sets are shared by all members of "quick_selects" so must not change
+#ifndef DBUG_OFF
+    DBUG_ASSERT(quick->head->read_set == save_read_set);
+    DBUG_ASSERT(quick->head->write_set == save_write_set);
+#endif
     /* All merged scans share the same record buffer in intersection. */
     quick->record= head->record[0];
   }
@@ -2327,8 +2342,8 @@ int QUICK_ROR_UNION_SELECT::reset()
   List_iterator_fast<QUICK_SELECT_I> it(quick_selects);
   while ((quick= it++))
   {
-    if (quick->reset())
-      DBUG_RETURN(1);
+    if ((error= quick->reset()))
+      DBUG_RETURN(error);
     if ((error= quick->get_next()))
     {
       if (error == HA_ERR_END_OF_FILE)
@@ -2339,10 +2354,10 @@ int QUICK_ROR_UNION_SELECT::reset()
     queue_insert(&queue, (uchar*)quick);
   }
 
-  if (head->file->ha_rnd_init_with_error(1))
+  if ((error= head->file->ha_rnd_init(1)))
   {
     DBUG_PRINT("error", ("ROR index_merge rnd_init call failed"));
-    DBUG_RETURN(1);
+    DBUG_RETURN(error);
   }
 
   DBUG_RETURN(0);
@@ -10610,7 +10625,10 @@ int read_keys_and_merge_scans(THD *thd,
     *unique_ptr= unique;
   }
   else
+  {
     unique->reset();
+    filesort_free_buffers(head, false);
+  }
 
   DBUG_ASSERT(file->ref_length == unique->get_size());
   DBUG_ASSERT(thd->variables.sortbuff_size == unique->get_max_in_memory_size());
@@ -10769,6 +10787,13 @@ int QUICK_INDEX_INTERSECT_SELECT::get_next()
     If a Clustered PK scan is present, it is used only to check if row
     satisfies its condition (and never used for row retrieval).
 
+    Locking: to ensure that exclusive locks are only set on records that
+    are included in the final result we must release the lock
+    on all rows we read but do not include in the final result. This
+    must be done on each index that reads the record and the lock
+    must be released using the same handler (the same quick object) as
+    used when reading the record.
+
   RETURN
    0     - Ok
    other - Error code if any error occurred.
@@ -10779,6 +10804,12 @@ int QUICK_ROR_INTERSECT_SELECT::get_next()
   List_iterator_fast<QUICK_SELECT_WITH_RECORD> quick_it(quick_selects);
   QUICK_SELECT_WITH_RECORD *qr;
   QUICK_RANGE_SELECT* quick;
+
+  /* quick that reads the given rowid first. This is needed in order
+  to be able to unlock the row using the same handler object that locked
+  it */
+  QUICK_RANGE_SELECT* quick_with_last_rowid;
+
   int error, cmp;
   uint last_rowid_count=0;
   DBUG_ENTER("QUICK_ROR_INTERSECT_SELECT::get_next");
@@ -10792,7 +10823,10 @@ int QUICK_ROR_INTERSECT_SELECT::get_next()
     if (cpk_quick)
     {
       while (!error && !cpk_quick->row_in_ranges())
+      {
+        quick->file->unlock_row(); /* row not in range; unlock */
         error= quick->get_next();
+      }
     }
     if (error)
       DBUG_RETURN(error);
@@ -10804,6 +10838,7 @@ int QUICK_ROR_INTERSECT_SELECT::get_next()
     quick->file->position(quick->record);
     memcpy(last_rowid, quick->file->ref, head->file->ref_length);
     last_rowid_count= 1;
+    quick_with_last_rowid= quick;
 
     while (last_rowid_count < quick_selects.elements)
     {
@@ -10817,9 +10852,17 @@ int QUICK_ROR_INTERSECT_SELECT::get_next()
       do
       {
         if ((error= quick->get_next()))
+        {
+          quick_with_last_rowid->file->unlock_row();
           DBUG_RETURN(error);
+        }
         quick->file->position(quick->record);
         cmp= head->file->cmp_ref(quick->file->ref, last_rowid);
+        if (cmp < 0)
+        {
+          /* This row is being skipped.  Release lock on it. */
+          quick->file->unlock_row();
+        }
       } while (cmp < 0);
 
       key_copy(qr->key_tuple, record, head->key_info + quick->index,
@@ -10833,13 +10876,19 @@ int QUICK_ROR_INTERSECT_SELECT::get_next()
         {
           while (!cpk_quick->row_in_ranges())
           {
+            quick->file->unlock_row(); /* row not in range; unlock */
             if ((error= quick->get_next()))
+            {
+              quick_with_last_rowid->file->unlock_row();
               DBUG_RETURN(error);
+            }
           }
           quick->file->position(quick->record);
         }
         memcpy(last_rowid, quick->file->ref, head->file->ref_length);
+        quick_with_last_rowid->file->unlock_row();
         last_rowid_count= 1;
+        quick_with_last_rowid= quick;
 
         //save the fields here
         key_copy(qr->key_tuple, record, head->key_info + quick->index,
@@ -10947,9 +10996,15 @@ int QUICK_RANGE_SELECT::reset()
   uchar *mrange_buff;
   int   error;
   HANDLER_BUFFER empty_buf;
+  MY_BITMAP * const save_read_set= head->read_set;
+  MY_BITMAP * const save_write_set= head->write_set;
   DBUG_ENTER("QUICK_RANGE_SELECT::reset");
   last_range= NULL;
   cur_range= (QUICK_RANGE**) ranges.buffer;
+  RANGE_SEQ_IF seq_funcs= {NULL, quick_range_seq_init, quick_range_seq_next, 0, 0};
+
+  if (in_ror_merged_scan)
+    head->column_bitmaps_set_no_signal(&column_bitmap, &column_bitmap);
   
   if (file->inited == handler::RND)
   {
@@ -10960,10 +11015,13 @@ int QUICK_RANGE_SELECT::reset()
 
   if (file->inited == handler::NONE)
   {
-    if (in_ror_merged_scan)
-      head->column_bitmaps_set_no_signal(&column_bitmap, &column_bitmap);
+    DBUG_EXECUTE_IF("bug14365043_2",
+                    DBUG_SET("+d,ha_index_init_fail"););
     if ((error= file->ha_index_init(index,1)))
-        DBUG_RETURN(error);
+    {
+        file->print_error(error, MYF(0));
+        goto err;
+    }
   }
 
   /* Allocate buffer if we need one but haven't allocated it yet */
@@ -10998,10 +11056,14 @@ int QUICK_RANGE_SELECT::reset()
   if (!mrr_buf_desc)
     empty_buf.buffer= empty_buf.buffer_end= empty_buf.end_of_used_area= NULL;
  
-  RANGE_SEQ_IF seq_funcs= {NULL, quick_range_seq_init, quick_range_seq_next, 0, 0};
   error= file->multi_range_read_init(&seq_funcs, (void*)this, ranges.elements,
                                      mrr_flags, mrr_buf_desc? mrr_buf_desc: 
                                                               &empty_buf);
+err:
+  /* Restore bitmaps set on entry */
+  if (in_ror_merged_scan)
+    head->column_bitmaps_set_no_signal(save_read_set, save_write_set);
+
   DBUG_RETURN(error);
 }
 
@@ -11024,6 +11086,9 @@ int QUICK_RANGE_SELECT::reset()
 int QUICK_RANGE_SELECT::get_next()
 {
   range_id_t dummy;
+  MY_BITMAP * const save_read_set= head->read_set;
+  MY_BITMAP * const save_write_set= head->write_set;
+
   DBUG_ENTER("QUICK_RANGE_SELECT::get_next");
   if (in_ror_merged_scan)
   {
@@ -13169,7 +13234,10 @@ int QUICK_GROUP_MIN_MAX_SELECT::reset(void)
     head->enable_keyread(); /* We need only the key attributes */
   }
   if ((result= file->ha_index_init(index,1)))
+  {
+    head->file->print_error(result, MYF(0));
     DBUG_RETURN(result);
+  }
   if (quick_prefix_select && quick_prefix_select->reset())
     DBUG_RETURN(1);
   result= file->ha_index_last(record);
@@ -13344,9 +13412,10 @@ int QUICK_GROUP_MIN_MAX_SELECT::next_min()
     */
     if (min_max_arg_part && min_max_arg_part->field->is_null())
     {
+      uchar *tmp_key_buff= (uchar*)my_alloca(max_used_key_length);
       /* Find the first subsequent record without NULL in the MIN/MAX field. */
-      key_copy(tmp_record, record, index_info, max_used_key_length);
-      result= file->ha_index_read_map(record, tmp_record,
+      key_copy(tmp_key_buff, record, index_info, max_used_key_length);
+      result= file->ha_index_read_map(record, tmp_key_buff,
                                       make_keypart_map(real_key_parts),
                                       HA_READ_AFTER_KEY);
       /*
@@ -13362,10 +13431,11 @@ int QUICK_GROUP_MIN_MAX_SELECT::next_min()
       if (!result)
       {
         if (key_cmp(index_info->key_part, group_prefix, real_prefix_len))
-          key_restore(record, tmp_record, index_info, 0);
+          key_restore(record, tmp_key_buff, index_info, 0);
       }
       else if (result == HA_ERR_KEY_NOT_FOUND || result == HA_ERR_END_OF_FILE)
         result= 0; /* There is a result in any case. */
+      my_afree(tmp_key_buff);
     }
   }