This won't be pushed either

mysql-test/t/handler.test:
  LIMIT tests
sql/mysql_priv.h:
  HANDLER ... LIMIT added
sql/sql_handler.cc:
  HANDLER ... LIMIT added
sql/sql_parse.cc:
  HANDLER ... LIMIT added
sql/sql_yacc.yy:
  HANDLER ... LIMIT added

5 files changed, 97 insertions, 73 deletions

diff --git a/mysql-test/t/handler.test b/mysql-test/t/handler.test
index a84a574aa29..9656a28abe0 100644
--- a/mysql-test/t/handler.test
+++ b/mysql-test/t/handler.test
@@ -45,5 +45,12 @@ handler t2 read a>(18);
 handler t2 read a<=(18);
 handler t2 read a<(18);
 
+handler t2 read a first limit 5;
+handler t2 read a next  limit 3;
+handler t2 read a prev  limit 10;
+
+handler t2 read a>=(16) limit 4;
+handler t2 read a last  limit 3;
+
 handler t2 close;
 drop table if exists t1;
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index 27fc5d02988..89bb69f6d92 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -385,7 +385,7 @@ int mysqld_show(THD *thd, const char *wild, show_var_st *variables);
 int mysql_ha_open(THD *thd, TABLE_LIST *tables);
 int mysql_ha_close(THD *thd, TABLE_LIST *tables);
 int mysql_ha_read(THD *, TABLE_LIST *,enum enum_ha_read_modes,
-                          char *,List<Item> *,enum ha_rkey_function);
+               char *,List<Item> *,enum ha_rkey_function,ha_rows,ha_rows);
 
 /* sql_base.cc */
 void set_item_name(Item *item,char *pos,uint length);
diff --git a/sql/sql_handler.cc b/sql/sql_handler.cc
index 17d5ea02e82..0d82fda8844 100644
--- a/sql/sql_handler.cc
+++ b/sql/sql_handler.cc
@@ -38,9 +38,13 @@ int mysql_ha_close(THD *thd, TABLE_LIST *tables)
   return 0;
 }
 
+static enum enum_ha_read_modes rkey_to_rnext[]= 
+    { RNEXT, RNEXT, RPREV, RNEXT, RPREV, RNEXT, RPREV };
+    
 int mysql_ha_read(THD *thd, TABLE_LIST *tables,
     enum enum_ha_read_modes mode, char *keyname, List<Item> *key_expr,
-    enum ha_rkey_function ha_rkey_mode)
+    enum ha_rkey_function ha_rkey_mode,
+    ha_rows select_limit,ha_rows offset_limit)
 {
   int err;
   TABLE *table=find_table_by_name(thd, tables->db, tables->name);
@@ -71,81 +75,93 @@ int mysql_ha_read(THD *thd, TABLE_LIST *tables,
   
   table->file->index_init(keyno);
   
-  switch(mode)
+  if (select_limit == thd->default_select_limit) select_limit=1;
+
+  select_limit+=offset_limit;  
+  for (uint num_rows=0; num_rows < select_limit; num_rows++)
   {
-    case RFIRST:
-       err=table->file->index_first(table->record[0]);
-       break;
-    case RLAST:
-       err=table->file->index_last(table->record[0]);
-       break;
-    case RNEXT:
-       err=table->file->index_next(table->record[0]);
-       break;
-    case RPREV:
-        err=table->file->index_prev(table->record[0]);
-        break;
-    case RKEY:
-      {
-        KEY *keyinfo=table->key_info+keyno;
-	uint key_len=0, i;
-	byte *key, *buf;
-	if (key_expr->elements > keyinfo->key_parts)
-	{
-	   my_printf_error(ER_TOO_MANY_KEY_PARTS,ER(ER_TOO_MANY_KEY_PARTS),
-	       MYF(0),keyinfo->key_parts);
-           return -1;
-	}
-	for (i=0; i < key_expr->elements; i++)
-	  key_len+=keyinfo->key_part[i].store_length;
-	if (!(key=sql_calloc(ALIGN_SIZE(key_len))))
-	{
-	  send_error(&thd->net,ER_OUTOFMEMORY);
-	  exit(-1); 
-	}
-        List_iterator<Item> it_ke(*key_expr);
-      	for (i=0, buf=key; i < key_expr->elements; i++)
+    switch(mode)
+    {
+      case RFIRST:
+	 err=table->file->index_first(table->record[0]);
+	 mode=RNEXT;
+	 break;
+      case RLAST:
+	 err=table->file->index_last(table->record[0]);
+	 mode=RPREV;
+	 break;
+      case RNEXT:
+	 err=table->file->index_next(table->record[0]);
+	 break;
+      case RPREV:
+	  err=table->file->index_prev(table->record[0]);
+	  break;
+      case RKEY:
 	{
-          uint maybe_null= test(keyinfo->key_part[i].null_bit);
-	  store_key_item ski=store_key_item(keyinfo->key_part[i].field,
-	      (char*)buf+maybe_null,maybe_null ? (char*) buf : 0,
-	      keyinfo->key_part[i].length, it_ke++);
-	  ski.copy();
-	  buf+=keyinfo->key_part[i].store_length;
+	  KEY *keyinfo=table->key_info+keyno;
+	  uint key_len=0, i;
+	  byte *key, *buf;
+	  if (key_expr->elements > keyinfo->key_parts)
+	  {
+	     my_printf_error(ER_TOO_MANY_KEY_PARTS,ER(ER_TOO_MANY_KEY_PARTS),
+		 MYF(0),keyinfo->key_parts);
+             return -1;
+	  }
+	  for (i=0; i < key_expr->elements; i++)
+	    key_len+=keyinfo->key_part[i].store_length;
+	  if (!(key=sql_calloc(ALIGN_SIZE(key_len))))
+	  {
+	    send_error(&thd->net,ER_OUTOFMEMORY);
+	    return -1; 
+	  }
+	  List_iterator<Item> it_ke(*key_expr);
+	  for (i=0, buf=key; i < key_expr->elements; i++)
+	  {
+            uint maybe_null= test(keyinfo->key_part[i].null_bit);
+	    store_key_item ski=store_key_item(keyinfo->key_part[i].field,
+		(char*)buf+maybe_null,maybe_null ? (char*) buf : 0,
+		keyinfo->key_part[i].length, it_ke++);
+	    ski.copy();
+	    buf+=keyinfo->key_part[i].store_length;
+	  }
+	  err=table->file->index_read(table->record[0],
+	      key,key_len,ha_rkey_mode);
+	  mode=rkey_to_rnext[(int)ha_rkey_mode];
+	  break;
 	}
-        err=table->file->index_read(table->record[0],
-	    key,key_len,ha_rkey_mode);
-        break;
-      }
-    default:
-	send_error(&thd->net,ER_ILLEGAL_HA);
-        exit(-1); 
-  }
-  
-  if (err && err != HA_ERR_KEY_NOT_FOUND && err != HA_ERR_END_OF_FILE)
-  {
-    sql_print_error("mysql_ha_read: Got error %d when reading table",
-                      err);
-    table->file->print_error(err,MYF(0));
-    return -1;
-  }
-  send_fields(thd,list,1);
-  if (!err)
-  {
-    String *packet = &thd->packet;
-    Item *item;
-    packet->length(0);
-    it.rewind();
-    while ((item=it++))
+      default:
+	  send_error(&thd->net,ER_ILLEGAL_HA);
+	  return -1; 
+    }
+
+    if (err && err != HA_ERR_KEY_NOT_FOUND && err != HA_ERR_END_OF_FILE)
+    {
+      sql_print_error("mysql_ha_read: Got error %d when reading table",
+                	err);
+      table->file->print_error(err,MYF(0));
+      return -1;
+    }
+    if (num_rows>=offset_limit)
     {
-      if (item->send(packet))
+      if (num_rows==offset_limit) send_fields(thd,list,1);
+      if (!err)
       {
-        packet->free();				// Free used
-        my_error(ER_OUT_OF_RESOURCES,MYF(0));
-        return -1;
+	String *packet = &thd->packet;
+	Item *item;
+	packet->length(0);
+	it.rewind();
+	while ((item=it++))
+	{
+	  if (item->send(packet))
+	  {
+	    packet->free();				// Free used
+	    my_error(ER_OUT_OF_RESOURCES,MYF(0));
+	    return -1;
+	  }
+	}
+	my_net_write(&thd->net, (char*)packet->ptr(), packet->length());
       }
     }
-    my_net_write(&thd->net, (char*)packet->ptr(), packet->length());
   }
   send_eof(&thd->net);
   return 0;
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 0789f2c07bf..72f163ec677 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -2010,7 +2010,8 @@ mysql_execute_command(void)
     if (check_db_used(thd,tables) || check_table_access(thd,SELECT_ACL, tables))
       goto error;
     res = mysql_ha_read(thd, tables, lex->ha_read_mode,
-	            lex->backup_dir, lex->insert_list, lex->ha_rkey_mode);
+	            lex->backup_dir, lex->insert_list, lex->ha_rkey_mode,
+	            lex->select_limit, lex->offset_limit);
     break;
 
   case SQLCOM_BEGIN:
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index a2de97534b7..9942be1506f 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -201,6 +201,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b,int *yystacksize);
 %token  LEVEL_SYM
 %token	LEX_HOSTNAME
 %token	LIKE
+%token	LIMIT
 %token	LINES
 %token	LOCAL_SYM
 %token	LOGS_SYM
@@ -312,7 +313,6 @@ bool my_yyoverflow(short **a, YYSTYPE **b,int *yystacksize);
 %token	FAST_SYM
 %token	FLOAT_SYM
 %token	INT_SYM
-%token	LIMIT
 %token	LONGBLOB
 %token	LONGTEXT
 %token	MEDIUMBLOB
@@ -2860,7 +2860,7 @@ handler:
 	  if (!add_table_to_list($2,0,0))
 	    YYABORT;
 	}
-	| HANDLER_SYM table_ident READ_SYM ident handler_read_function
+	| HANDLER_SYM table_ident READ_SYM ident handler_read_function limit_clause
 	{
 	  Lex->sql_command = SQLCOM_HA_READ;
 	  Lex->backup_dir= $4.str;