BUG#121915: Merge 4.1->5.0

mysql-test/r/update.result:
  Auto merged
mysql-test/t/update.test:
  Auto merged
sql/mysql_priv.h:
  Auto merged
sql/mysqld.cc:
  Auto merged
sql/records.cc:
  Auto merged
sql/repl_failsafe.cc:
  Auto merged
sql/slave.cc:
  Auto merged
sql/structs.h:
  Auto merged

13 files changed, 344 insertions, 20 deletions

diff --git a/mysql-test/r/update.result b/mysql-test/r/update.result
index 9ca6370dff6..d2aa051c299 100644
--- a/mysql-test/r/update.result
+++ b/mysql-test/r/update.result
@@ -263,3 +263,62 @@ test
 delete from t1 where count(*)=1;
 ERROR HY000: Invalid use of group function
 drop table t1;
+create table t1 ( a int, index (a) );
+insert into t1 values (0),(0),(0),(0),(0),(0),(0),(0);
+flush status;
+select a from t1 order by a limit 1;
+a
+0
+show status like 'handler_read%';
+Variable_name	Value
+Handler_read_first	1
+Handler_read_key	0
+Handler_read_next	0
+Handler_read_prev	0
+Handler_read_rnd	0
+Handler_read_rnd_next	0
+flush status;
+update t1 set a=unix_timestamp() order by a limit 1;
+show status like 'handler_read%';
+Variable_name	Value
+Handler_read_first	1
+Handler_read_key	0
+Handler_read_next	0
+Handler_read_prev	0
+Handler_read_rnd	1
+Handler_read_rnd_next	0
+flush status;
+delete from t1 order by a limit 1;
+show status like 'handler_read%';
+Variable_name	Value
+Handler_read_first	1
+Handler_read_key	0
+Handler_read_next	0
+Handler_read_prev	0
+Handler_read_rnd	0
+Handler_read_rnd_next	0
+flush status;
+delete from t1 order by a desc limit 1;
+show status like 'handler_read%';
+Variable_name	Value
+Handler_read_first	0
+Handler_read_key	0
+Handler_read_next	0
+Handler_read_prev	0
+Handler_read_rnd	1
+Handler_read_rnd_next	9
+alter table t1 disable keys;
+flush status;
+delete from t1 order by a limit 1;
+show status like 'handler_read%';
+Variable_name	Value
+Handler_read_first	0
+Handler_read_key	0
+Handler_read_next	0
+Handler_read_prev	0
+Handler_read_rnd	1
+Handler_read_rnd_next	9
+select count(*) from t1;
+count(*)
+5
+drop table t1;
diff --git a/mysql-test/t/update.test b/mysql-test/t/update.test
index 3822d1fe4e7..13ced76079f 100644
--- a/mysql-test/t/update.test
+++ b/mysql-test/t/update.test
@@ -227,4 +227,33 @@ select DATABASE();
 delete from t1 where count(*)=1;
 drop table t1;
 
+# BUG#12915: Optimize "DELETE|UPDATE ... ORDER BY ... LIMIT n" to use an index
+create table t1 ( a int, index (a) );
+insert into t1 values (0),(0),(0),(0),(0),(0),(0),(0);
+
+flush status;
+select a from t1 order by a limit 1;
+show status like 'handler_read%';
+
+flush status;
+update t1 set a=unix_timestamp() order by a limit 1;
+show status like 'handler_read%';
+
+flush status;
+delete from t1 order by a limit 1;
+show status like 'handler_read%';
+
+flush status;
+delete from t1 order by a desc limit 1;
+show status like 'handler_read%';
+
+alter table t1 disable keys;
+
+flush status;
+delete from t1 order by a limit 1;
+show status like 'handler_read%';
+
+select count(*) from t1;
+
+drop table t1;
 # End of 4.1 tests
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index 27eb6556d66..e33ac05e293 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -1334,6 +1334,8 @@ void change_byte(byte *,uint,char,char);
 void init_read_record(READ_RECORD *info, THD *thd, TABLE *reg_form,
 		      SQL_SELECT *select,
 		      int use_record_cache, bool print_errors);
+void init_read_record_idx(READ_RECORD *info, THD *thd, TABLE *table, 
+                          bool print_error, uint idx);
 void end_read_record(READ_RECORD *info);
 ha_rows filesort(THD *thd, TABLE *form,struct st_sort_field *sortorder,
 		 uint s_length, SQL_SELECT *select,
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 4c46f601ee1..9e1fa4bb5ee 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -3674,7 +3674,6 @@ static void bootstrap(FILE *file)
 
   THD *thd= new THD;
   thd->bootstrap=1;
-  thd->client_capabilities=0;
   my_net_init(&thd->net,(st_vio*) 0);
   thd->max_client_packet_length= thd->net.max_packet;
   thd->security_ctx->master_access= ~(ulong)0;
diff --git a/sql/opt_range.cc b/sql/opt_range.cc
index 1a6d97caa6a..7e1821b9953 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -1363,6 +1363,95 @@ SEL_ARG *SEL_ARG::clone_tree()
 
 
 /*
+  Find the best index to retrieve first N records in given order
+
+  SYNOPSIS
+    get_index_for_order()
+      table  Table to be accessed
+      order  Required ordering
+      limit  Number of records that will be retrieved
+
+  DESCRIPTION
+    Find the best index that allows to retrieve first #limit records in the 
+    given order cheaper then one would retrieve them using full table scan.
+
+  IMPLEMENTATION
+    Run through all table indexes and find the shortest index that allows
+    records to be retrieved in given order. We look for the shortest index
+    as we will have fewer index pages to read with it.
+
+    This function is used only by UPDATE/DELETE, so we take into account how
+    the UPDATE/DELETE code will work:
+     * index can only be scanned in forward direction
+     * HA_EXTRA_KEYREAD will not be used
+    Perhaps these assumptions could be relaxed
+
+  RETURN
+    index number
+    MAX_KEY if no such index was found.
+*/
+
+uint get_index_for_order(TABLE *table, ORDER *order, ha_rows limit)
+{
+  uint idx;
+  uint match_key= MAX_KEY, match_key_len= MAX_KEY_LENGTH + 1;
+  ORDER *ord;
+  
+  for (ord= order; ord; ord= ord->next)
+    if (!ord->asc)
+      return MAX_KEY;
+
+  for (idx= 0; idx < table->keys; idx++)
+  {
+    if (!(table->keys_in_use_for_query.is_set(idx)))
+      continue;
+    KEY_PART_INFO *keyinfo= table->key_info[idx].key_part;
+    uint partno= 0;
+    
+    /* 
+      The below check is sufficient considering we now have either BTREE 
+      indexes (records are returned in order for any index prefix) or HASH 
+      indexes (records are not returned in order for any index prefix).
+    */
+    if (!(table->file->index_flags(idx, 0, 1) & HA_READ_ORDER))
+      continue;
+    for (ord= order; ord; ord= ord->next, partno++)
+    {
+      Item *item= order->item[0];
+      if (!(item->type() == Item::FIELD_ITEM &&
+           ((Item_field*)item)->field->eq(keyinfo[partno].field)))
+        break;
+    }
+    
+    if (!ord && table->key_info[idx].key_length < match_key_len)
+    {
+      /* 
+        Ok, the ordering is compatible and this key is shorter then
+        previous match (we want shorter keys as we'll have to read fewer
+        index pages for the same number of records)
+      */
+      match_key= idx;
+      match_key_len= table->key_info[idx].key_length;
+    }
+  }
+
+  if (match_key != MAX_KEY)
+  {
+    /* 
+      Found an index that allows records to be retrieved in the requested 
+      order. Now we'll check if using the index is cheaper then doing a table
+      scan.
+    */
+    double full_scan_time= table->file->scan_time();
+    double index_scan_time= table->file->read_time(match_key, 1, limit);
+    if (index_scan_time > full_scan_time)
+      match_key= MAX_KEY;
+  }
+  return match_key;
+}
+
+
+/*
   Table rows retrieval plan. Range optimizer creates QUICK_SELECT_I-derived
   objects from table read plans.
 */
diff --git a/sql/opt_range.h b/sql/opt_range.h
index 37d77033c8d..f84058f3b64 100644
--- a/sql/opt_range.h
+++ b/sql/opt_range.h
@@ -716,4 +716,6 @@ public:
 QUICK_RANGE_SELECT *get_quick_select_for_ref(THD *thd, TABLE *table,
                                              struct st_table_ref *ref,
                                              ha_rows records);
+uint get_index_for_order(TABLE *table, ORDER *order, ha_rows limit);
+
 #endif
diff --git a/sql/records.cc b/sql/records.cc
index 9b05dc3e291..52c3dbdb798 100644
--- a/sql/records.cc
+++ b/sql/records.cc
@@ -29,7 +29,59 @@ static int rr_from_cache(READ_RECORD *info);
 static int init_rr_cache(THD *thd, READ_RECORD *info);
 static int rr_cmp(uchar *a,uchar *b);
 
-	/* init struct for read with info->read_record */
+static int rr_index(READ_RECORD *info);
+
+
+
+/*
+  Initialize READ_RECORD structure to perform full index scan
+
+  SYNOPSIS 
+    init_read_record_idx()
+      info         READ_RECORD structure to initialize.
+      thd          Thread handle
+      table        Table to be accessed 
+      print_error  If true, call table->file->print_error() if an error
+                   occurs (except for end-of-records error)
+      idx          index to scan
+  
+  DESCRIPTION
+    Initialize READ_RECORD structure to perform full index scan (in forward
+    direction) using read_record.read_record() interface.
+    
+    This function has been added at late stage and is used only by
+    UPDATE/DELETE. Other statements perform index scans using
+    join_read_first/next functions.
+*/
+
+void init_read_record_idx(READ_RECORD *info, THD *thd, TABLE *table,
+                          bool print_error, uint idx)
+{
+  bzero((char*) info,sizeof(*info));
+  info->thd=thd;
+  info->table=table;
+  info->file= table->file;
+  info->forms= &info->table;		/* Only one table */
+    
+  info->record= table->record[0];
+  info->ref_length= table->file->ref_length;
+
+  info->select=NULL;
+  info->print_error=print_error;
+  info->ignore_not_found_rows= 0;
+  table->status=0;			/* And it's always found */
+
+  if (!table->file->inited)
+  {
+    table->file->ha_index_init(idx);
+    table->file->extra(HA_EXTRA_RETRIEVE_PRIMARY_KEY);
+  }
+  info->read_record= rr_index;
+  info->first= TRUE;
+}
+
+
+/* init struct for read with info->read_record */
 
 void init_read_record(READ_RECORD *info,THD *thd, TABLE *table,
 		      SQL_SELECT *select,
@@ -187,6 +239,60 @@ static int rr_quick(READ_RECORD *info)
 }
 
 
+/*
+  A READ_RECORD::read_record implementation that reads index sequentially
+
+  SYNOPSIS
+    rr_index()
+      info  Scan info
+  
+  DESCRIPTION
+    Read the next index record (in forward direction) and translate return
+    value.
+    
+  RETURN
+    0   Ok
+    -1  End of records 
+    1   Error   
+*/
+
+static int rr_index(READ_RECORD *info)
+{
+  int tmp;
+  while (1)
+  {
+    if (info->first)
+    {
+      info->first= FALSE;
+      tmp= info->file->index_first(info->record);
+    }
+    else
+      tmp= info->file->index_next(info->record);
+    
+    if (!tmp)
+      break;
+    if (info->thd->killed)
+    {
+      my_error(ER_SERVER_SHUTDOWN,MYF(0));
+      return 1;
+    }
+    if (tmp != HA_ERR_RECORD_DELETED)
+    {
+      if (tmp == HA_ERR_END_OF_FILE)
+        tmp= -1;
+      else
+      {
+        if (info->print_error)
+          info->table->file->print_error(tmp,MYF(0));
+        if (tmp < 0)                            // Fix negative BDB errno
+          tmp=1;
+      }
+      break;
+    }
+  }
+  return tmp;
+}
+
 static int rr_sequential(READ_RECORD *info)
 {
   int tmp;
diff --git a/sql/repl_failsafe.cc b/sql/repl_failsafe.cc
index 792251e82c0..2f00e43deec 100644
--- a/sql/repl_failsafe.cc
+++ b/sql/repl_failsafe.cc
@@ -67,7 +67,6 @@ static int init_failsafe_rpl_thread(THD* thd)
   */
   thd->system_thread = thd->bootstrap = 1;
   thd->security_ctx->skip_grants();
-  thd->client_capabilities = 0;
   my_net_init(&thd->net, 0);
   thd->net.read_timeout = slave_net_timeout;
   thd->max_client_packet_length=thd->net.max_packet;
diff --git a/sql/slave.cc b/sql/slave.cc
index 603ba8a8155..f785bdcd4d7 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -2869,7 +2869,6 @@ static int init_slave_thread(THD* thd, SLAVE_THD_TYPE thd_type)
   thd->system_thread = (thd_type == SLAVE_THD_SQL) ?
     SYSTEM_THREAD_SLAVE_SQL : SYSTEM_THREAD_SLAVE_IO; 
   thd->security_ctx->skip_grants();
-  thd->client_capabilities = 0;
   my_net_init(&thd->net, 0);
   thd->net.read_timeout = slave_net_timeout;
   thd->slave_thread = 1;
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index abb3cf666b0..609156ef5a8 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -218,6 +218,7 @@ THD::THD()
 #ifndef EMBEDDED_LIBRARY
   net.vio=0;
 #endif
+  client_capabilities= 0;                       // minimalistic client
   net.last_error[0]=0;                          // If error on boot
   net.query_cache_query=0;                      // If error on boot
   ull=0;
diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index d9734b7cae8..1690f96a3d4 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -37,6 +37,7 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
   bool          using_limit=limit != HA_POS_ERROR;
   bool		transactional_table, safe_update, const_cond;
   ha_rows	deleted;
+  uint usable_index= MAX_KEY;
   SELECT_LEX   *select_lex= &thd->lex->select_lex;
   DBUG_ENTER("mysql_delete");
 
@@ -140,27 +141,42 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
     tables.table = table;
     tables.alias = table_list->alias;
 
-    table->sort.io_cache = (IO_CACHE *) my_malloc(sizeof(IO_CACHE),
-                                             MYF(MY_FAE | MY_ZEROFILL));
       if (select_lex->setup_ref_array(thd, order->elements) ||
 	  setup_order(thd, select_lex->ref_pointer_array, &tables,
-		      fields, all_fields, (ORDER*) order->first) ||
-	  !(sortorder=make_unireg_sortorder((ORDER*) order->first, &length)) ||
+                    fields, all_fields, (ORDER*) order->first))
+    {
+      delete select;
+      free_underlaid_joins(thd, &thd->lex->select_lex);
+      DBUG_RETURN(-1);			// This will force out message
+    }
+    
+    if (!select && limit != HA_POS_ERROR)
+      usable_index= get_index_for_order(table, (ORDER*)(order->first), limit);
+
+    if (usable_index == MAX_KEY)
+    {
+      table->sort.io_cache= (IO_CACHE *) my_malloc(sizeof(IO_CACHE),
+                                                   MYF(MY_FAE | MY_ZEROFILL));
+    
+      if ( !(sortorder=make_unireg_sortorder((ORDER*) order->first, &length)) ||
 	  (table->sort.found_records = filesort(thd, table, sortorder, length,
 					   select, HA_POS_ERROR,
 					   &examined_rows))
 	  == HA_POS_ERROR)
-    {
+      {
+        delete select;
+        free_underlaid_joins(thd, &thd->lex->select_lex);
+        DBUG_RETURN(-1);			// This will force out message
+      }
+      /*
+        Filesort has already found and selected the rows we want to delete,
+        so we don't need the where clause
+      */
       delete select;
       free_underlaid_joins(thd, select_lex);
+      select= 0;
       DBUG_RETURN(TRUE);
     }
-    /*
-      Filesort has already found and selected the rows we want to delete,
-      so we don't need the where clause
-    */
-    delete select;
-    select= 0;
   }
 
   /* If quick select is used, initialize it before retrieving rows. */
@@ -170,7 +186,11 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
     free_underlaid_joins(thd, select_lex);
     DBUG_RETURN(TRUE);
   }
-  init_read_record(&info,thd,table,select,1,1);
+  if (usable_index==MAX_KEY)
+    init_read_record(&info,thd,table,select,1,1);
+  else
+    init_read_record_idx(&info, thd, table, 1, usable_index);
+
   deleted=0L;
   init_ftfuncs(thd, select_lex, 1);
   thd->proc_info="updating";
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index 9f002d6e3ca..a8e21177338 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -122,7 +122,8 @@ int mysql_update(THD *thd,
   bool		used_key_is_modified, transactional_table;
   int           res;
   int		error=0;
-  uint		used_index;
+  uint		used_index= MAX_KEY;
+  bool          need_sort= TRUE;
 #ifndef NO_EMBEDDED_ACCESS_CHECKS
   uint		want_privilege;
 #endif
@@ -241,6 +242,11 @@ int mysql_update(THD *thd,
     send_ok(thd);				// No matching records
     DBUG_RETURN(0);
   }
+  if (!select && limit != HA_POS_ERROR)
+  {
+    if (MAX_KEY != (used_index= get_index_for_order(table, order, limit)))
+      need_sort= FALSE;
+  }
   /* If running in safe sql mode, don't allow updates without keys */
   if (table->quick_keys.is_clear_all())
   {
@@ -261,6 +267,10 @@ int mysql_update(THD *thd,
     used_key_is_modified= (!select->quick->unique_key_range() &&
                           select->quick->check_if_keys_used(&fields));
   }
+  else if (used_index != MAX_KEY)
+  {
+    used_key_is_modified= check_if_key_used(table, used_index, fields);
+  }
   else if ((used_index=table->file->key_used_on_scan) < MAX_KEY)
     used_key_is_modified=check_if_key_used(table, used_index, fields);
   else
@@ -276,10 +286,11 @@ int mysql_update(THD *thd,
     if (used_index < MAX_KEY && old_used_keys.is_set(used_index))
     {
       table->key_read=1;
-      table->file->extra(HA_EXTRA_KEYREAD);
+      table->file->extra(HA_EXTRA_KEYREAD); //todo: psergey: check
     }
 
-    if (order)
+    /* note: can actually avoid sorting below.. */
+    if (order && need_sort)
     {
       /*
 	Doing an ORDER BY;  Let filesort find and sort the rows we are going
@@ -323,7 +334,10 @@ int mysql_update(THD *thd,
       /* If quick select is used, initialize it before retrieving rows. */
       if (select && select->quick && select->quick->reset())
         goto err;
-      init_read_record(&info,thd,table,select,0,1);
+      if (used_index == MAX_KEY)
+        init_read_record(&info,thd,table,select,0,1);
+      else
+        init_read_record_idx(&info, thd, table, 1, used_index);
 
       thd->proc_info="Searching rows for update";
       uint tmp_limit= limit;
@@ -352,6 +366,10 @@ int mysql_update(THD *thd,
 	error= 1;				// Aborted
       limit= tmp_limit;
       end_read_record(&info);
+
+      /* if we got here we must not use index in the main update loop below */
+      used_index= MAX_KEY;
+      
       /* Change select to use tempfile */
       if (select)
       {
diff --git a/sql/structs.h b/sql/structs.h
index 21aaf584925..92f09385d6e 100644
--- a/sql/structs.h
+++ b/sql/structs.h
@@ -135,6 +135,7 @@ typedef struct st_read_record {			/* Parameter to read_record */
   byte	*cache,*cache_pos,*cache_end,*read_positions;
   IO_CACHE *io_cache;
   bool print_error, ignore_not_found_rows;
+  bool first; /* used only with rr_index_read */
 } READ_RECORD;