This won't be pushed either

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;

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);

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,13 +75,20 @@ int mysql_ha_read(THD *thd, TABLE_LIST *tables,
   
   table->file->index_init(keyno);
   
+  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++)
+  {
     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]);
@@ -101,7 +112,7 @@ int mysql_ha_read(THD *thd, TABLE_LIST *tables,
 	  if (!(key=sql_calloc(ALIGN_SIZE(key_len))))
 	  {
 	    send_error(&thd->net,ER_OUTOFMEMORY);
-	  exit(-1); 
+	    return -1; 
 	  }
 	  List_iterator<Item> it_ke(*key_expr);
 	  for (i=0, buf=key; i < key_expr->elements; i++)
@@ -115,11 +126,12 @@ int mysql_ha_read(THD *thd, TABLE_LIST *tables,
 	  }
 	  err=table->file->index_read(table->record[0],
 	      key,key_len,ha_rkey_mode);
+	  mode=rkey_to_rnext[(int)ha_rkey_mode];
 	  break;
 	}
       default:
 	  send_error(&thd->net,ER_ILLEGAL_HA);
-        exit(-1); 
+	  return -1; 
     }
 
     if (err && err != HA_ERR_KEY_NOT_FOUND && err != HA_ERR_END_OF_FILE)
@@ -129,7 +141,9 @@ int mysql_ha_read(THD *thd, TABLE_LIST *tables,
       table->file->print_error(err,MYF(0));
       return -1;
     }
-  send_fields(thd,list,1);
+    if (num_rows>=offset_limit)
+    {
+      if (num_rows==offset_limit) send_fields(thd,list,1);
       if (!err)
       {
 	String *packet = &thd->packet;
@@ -147,6 +161,8 @@ int mysql_ha_read(THD *thd, TABLE_LIST *tables,
 	}
 	my_net_write(&thd->net, (char*)packet->ptr(), packet->length());
       }
+    }
+  }
   send_eof(&thd->net);
   return 0;
 }

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:

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;