Backport into MariaDB-5.2 the following:

WL#2474 "Multi Range Read: Change the default MRR implementation to implement new MRR interface"
WL#2475 "Batched range read functions for MyISAM/InnoDb"
        "Index condition pushdown for MyISAM/InnoDB"
Igor's fix from sp1r-igor@olga.mysql.com-20080330055902-07614:
  There could be observed the following problems:
  1. EXPLAIN did not mention pushdown conditions from on expressions in the 
  'extra' column.  As a result if a query had no where conditions pushed 
  down to a table, but had on conditions pushed to this table the 'extra' 
  column in the EXPLAIN for the table missed 'using where'.
  2. Conditions for ref access were not eliminated from on expressions 
  though such conditions were eliminated from the where condition.

21 files changed, 809 insertions, 129 deletions

diff --git a/storage/maria/ha_maria.cc b/storage/maria/ha_maria.cc
index 1b583353126..772989109d0 100644
--- a/storage/maria/ha_maria.cc
+++ b/storage/maria/ha_maria.cc
@@ -2020,6 +2020,21 @@ int ha_maria::delete_row(const uchar * buf)
   return maria_delete(file, buf);
 }
 
+C_MODE_START
+
+my_bool index_cond_func_maria(void *arg)
+{
+  ha_maria *h= (ha_maria*)arg;
+  /*if (h->in_range_read)*/
+  if (h->end_range)
+  {
+    if (h->compare_key2(h->end_range) > 0)
+      return 2; /* caller should return HA_ERR_END_OF_FILE already */
+  }
+  return (my_bool)h->pushed_idx_cond->val_int();
+}
+
+C_MODE_END
 
 int ha_maria::index_read_map(uchar * buf, const uchar * key,
 			     key_part_map keypart_map,
@@ -2117,6 +2132,25 @@ int ha_maria::index_next_same(uchar * buf,
 }
 
 
+int ha_maria::index_init(uint idx, bool sorted)
+{ 
+  active_index=idx;
+  if (pushed_idx_cond_keyno == idx)
+    ma_set_index_cond_func(file, index_cond_func_maria, this);
+  return 0; 
+}
+
+
+int ha_maria::index_end()
+{
+  active_index=MAX_KEY;
+  ma_set_index_cond_func(file, NULL, 0);
+  in_range_check_pushed_down= FALSE;
+  ds_mrr.dsmrr_close();
+  return 0; 
+}
+
+
 int ha_maria::rnd_init(bool scan)
 {
   if (scan)
@@ -2201,6 +2235,7 @@ int ha_maria::info(uint flag, my_bool lock_table_share)
     ref_length= maria_info.reflength;
     share->db_options_in_use= maria_info.options;
     stats.block_size= maria_block_size;
+    stats.mrr_length_per_rec= maria_info.reflength + 8; // 8 = max(sizeof(void *))
 
     /* Update share */
     if (lock_table_share)
@@ -2275,6 +2310,10 @@ int ha_maria::extra(enum ha_extra_function operation)
 
 int ha_maria::reset(void)
 {
+  pushed_idx_cond= NULL;
+  pushed_idx_cond_keyno= MAX_KEY;
+  ma_set_index_cond_func(file, NULL, 0);
+  ds_mrr.dsmrr_close();
   return maria_reset(file);
 }
 
@@ -3333,6 +3372,64 @@ static SHOW_VAR status_variables[]= {
   {NullS, NullS, SHOW_LONG}
 };
 
+/****************************************************************************
+ * Maria MRR implementation: use DS-MRR
+ ***************************************************************************/
+
+int ha_maria::multi_range_read_init(RANGE_SEQ_IF *seq, void *seq_init_param,
+                                     uint n_ranges, uint mode, 
+                                     HANDLER_BUFFER *buf)
+{
+  return ds_mrr.dsmrr_init(this, seq, seq_init_param, n_ranges, mode, buf);
+}
+
+int ha_maria::multi_range_read_next(char **range_info)
+{
+  return ds_mrr.dsmrr_next(range_info);
+}
+
+ha_rows ha_maria::multi_range_read_info_const(uint keyno, RANGE_SEQ_IF *seq,
+                                               void *seq_init_param, 
+                                               uint n_ranges, uint *bufsz,
+                                               uint *flags, COST_VECT *cost)
+{
+  /*
+    This call is here because there is no location where this->table would
+    already be known.
+    TODO: consider moving it into some per-query initialization call.
+  */
+  ds_mrr.init(this, table);
+  return ds_mrr.dsmrr_info_const(keyno, seq, seq_init_param, n_ranges, bufsz,
+                                 flags, cost);
+}
+
+ha_rows ha_maria::multi_range_read_info(uint keyno, uint n_ranges, uint keys,
+                                        uint *bufsz, uint *flags, 
+                                        COST_VECT *cost)
+{
+  ds_mrr.init(this, table);
+  return ds_mrr.dsmrr_info(keyno, n_ranges, keys, bufsz, flags, cost);
+}
+
+/* MyISAM MRR implementation ends */
+
+
+/* Index condition pushdown implementation*/
+
+
+Item *ha_maria::idx_cond_push(uint keyno_arg, Item* idx_cond_arg)
+{
+  pushed_idx_cond_keyno= keyno_arg;
+  pushed_idx_cond= idx_cond_arg;
+  in_range_check_pushed_down= TRUE;
+  if (active_index == pushed_idx_cond_keyno)
+    ma_set_index_cond_func(file, index_cond_func_maria, this);
+  return NULL;
+}
+
+
+
+
 struct st_mysql_storage_engine maria_storage_engine=
 { MYSQL_HANDLERTON_INTERFACE_VERSION };
 
diff --git a/storage/maria/ha_maria.h b/storage/maria/ha_maria.h
index 23fb74f1c27..c295527c2dc 100644
--- a/storage/maria/ha_maria.h
+++ b/storage/maria/ha_maria.h
@@ -28,6 +28,10 @@
 #define HA_RECOVER_FORCE        4       /* Recover even if we loose rows */
 #define HA_RECOVER_QUICK        8       /* Don't check rows in data file */
 
+C_MODE_START
+my_bool index_cond_func_maria(void *arg);
+C_MODE_END
+
 extern ulong maria_sort_buffer_size;
 extern TYPELIB maria_recover_typelib;
 extern ulong maria_recover_options;
@@ -62,7 +66,7 @@ public:
   {
     return ((table_share->key_info[inx].algorithm == HA_KEY_ALG_FULLTEXT) ?
             0 : HA_READ_NEXT | HA_READ_PREV | HA_READ_RANGE |
-            HA_READ_ORDER | HA_KEYREAD_ONLY);
+            HA_READ_ORDER | HA_KEYREAD_ONLY | HA_DO_INDEX_COND_PUSHDOWN);
   }
   uint max_supported_keys() const
   { return MARIA_MAX_KEY; }
@@ -104,6 +108,8 @@ public:
                                 key->charset(), table->record[0]);
   }
   int ft_read(uchar * buf);
+  int index_init(uint idx, bool sorted);
+  int index_end();
   int rnd_init(bool scan);
   int rnd_end(void);
   int rnd_next(uchar * buf);
@@ -164,4 +170,22 @@ public:
     return file;
   }
   static int implicit_commit(THD *thd, bool new_trn);
+  /**
+   * Multi Range Read interface
+   */
+  int multi_range_read_init(RANGE_SEQ_IF *seq, void *seq_init_param,
+                            uint n_ranges, uint mode, HANDLER_BUFFER *buf);
+  int multi_range_read_next(char **range_info);
+  ha_rows multi_range_read_info_const(uint keyno, RANGE_SEQ_IF *seq,
+                                      void *seq_init_param, 
+                                      uint n_ranges, uint *bufsz,
+                                      uint *flags, COST_VECT *cost);
+  ha_rows multi_range_read_info(uint keyno, uint n_ranges, uint keys,
+                                uint *bufsz, uint *flags, COST_VECT *cost);
+  
+  /* Index condition pushdown implementation */
+  Item *idx_cond_push(uint keyno, Item* idx_cond);
+private:
+  DsMrr_impl ds_mrr;
+  friend my_bool index_cond_func_maria(void *arg);
 };
diff --git a/storage/maria/ma_extra.c b/storage/maria/ma_extra.c
index 691e55b2a2a..4499cca2885 100644
--- a/storage/maria/ma_extra.c
+++ b/storage/maria/ma_extra.c
@@ -474,6 +474,14 @@ int maria_extra(MARIA_HA *info, enum ha_extra_function function,
 } /* maria_extra */
 
 
+void ma_set_index_cond_func(MARIA_HA *info, index_cond_func_t func,
+                            void *func_arg)
+{
+  info->index_cond_func= func;
+  info->index_cond_func_arg= func_arg;
+}
+
+
 /*
   Start/Stop Inserting Duplicates Into a Table, WL#1648.
 */
@@ -620,3 +628,4 @@ int _ma_flush_table_files(MARIA_HA *info, uint flush_data_or_index,
   maria_mark_crashed(info);
   return 1;
 }
+
diff --git a/storage/maria/ma_key.c b/storage/maria/ma_key.c
index 08702a9109e..c1067873c0e 100644
--- a/storage/maria/ma_key.c
+++ b/storage/maria/ma_key.c
@@ -26,7 +26,8 @@
 
 #define CHECK_KEYS                              /* Enable safety checks */
 
-static int _ma_put_key_in_record(MARIA_HA *info,uint keynr,uchar *record);
+static int _ma_put_key_in_record(MARIA_HA *info, uint keynr,
+                                 my_bool unpack_blobs, uchar *record);
 
 #define FIX_LENGTH(cs, pos, length, char_length)                            \
             do {                                                            \
@@ -476,6 +477,9 @@ void _ma_copy_key(MARIA_KEY *to, const MARIA_KEY *from)
     _ma_put_key_in_record()
     info		MARIA handler
     keynr		Key number that was used
+    unpack_blobs        TRUE  <=> Unpack blob columns
+                        FALSE <=> Skip them. This is used by index condition 
+                                  pushdown check function
     record 		Store key here
 
     Last read key is in info->lastkey
@@ -489,7 +493,7 @@ void _ma_copy_key(MARIA_KEY *to, const MARIA_KEY *from)
 */
 
 static int _ma_put_key_in_record(register MARIA_HA *info, uint keynr,
-				 uchar *record)
+				 my_bool unpack_blobs, uchar *record)
 {
   reg2 uchar *key;
   uchar *pos,*key_end;
@@ -582,16 +586,19 @@ static int _ma_put_key_in_record(register MARIA_HA *info, uint keynr,
       if (length > keyseg->length || key+length > key_end)
 	goto err;
 #endif
-      memcpy(record+keyseg->start+keyseg->bit_start,
-	     (char*) &blob_ptr,sizeof(char*));
-      memcpy(blob_ptr,key,length);
-      blob_ptr+=length;
+      if (unpack_blobs)
+      {
+        memcpy(record+keyseg->start+keyseg->bit_start,
+               (char*) &blob_ptr,sizeof(char*));
+        memcpy(blob_ptr,key,length);
+        blob_ptr+=length;
 
-      /* The above changed info->lastkey2. Inform maria_rnext_same(). */
-      info->update&= ~HA_STATE_RNEXT_SAME;
+        /* The above changed info->lastkey2. Inform maria_rnext_same(). */
+        info->update&= ~HA_STATE_RNEXT_SAME;
 
-      _ma_store_blob_length(record+keyseg->start,
-			    (uint) keyseg->bit_start,length);
+        _ma_store_blob_length(record+keyseg->start,
+                              (uint) keyseg->bit_start,length);
+      }
       key+=length;
     }
     else if (keyseg->flag & HA_SWAP_KEY)
@@ -621,6 +628,7 @@ static int _ma_put_key_in_record(register MARIA_HA *info, uint keynr,
   DBUG_RETURN(0);
 
 err:
+  DBUG_PRINT("info",("error"));
   DBUG_RETURN(1);				/* Crashed row */
 } /* _ma_put_key_in_record */
 
@@ -634,7 +642,7 @@ int _ma_read_key_record(MARIA_HA *info, uchar *buf, MARIA_RECORD_POS filepos)
   {
     if (info->lastinx >= 0)
     {				/* Read only key */
-      if (_ma_put_key_in_record(info,(uint) info->lastinx,buf))
+      if (_ma_put_key_in_record(info, (uint)info->lastinx, TRUE, buf))
       {
         maria_print_error(info->s, HA_ERR_CRASHED);
 	my_errno=HA_ERR_CRASHED;
@@ -649,6 +657,40 @@ int _ma_read_key_record(MARIA_HA *info, uchar *buf, MARIA_RECORD_POS filepos)
 }
 
 
+
+/*
+  Save current key tuple to record and call index condition check function
+
+  SYNOPSIS
+    ma_check_index_cond()
+      info    MyISAM handler
+      keynr   Index we're running a scan on
+      record  Record buffer to use (it is assumed that index check function 
+              will look for column values there)
+
+  RETURN
+    -1  Error 
+    0   Index condition is not satisfied, continue scanning
+    1   Index condition is satisfied
+    2   Index condition is not satisfied, end the scan. 
+*/
+
+int ma_check_index_cond(register MARIA_HA *info, uint keynr, uchar *record)
+{
+  if (info->index_cond_func)
+  {
+    if (_ma_put_key_in_record(info, keynr, FALSE, record))
+    {
+      maria_print_error(info->s, HA_ERR_CRASHED);
+      my_errno=HA_ERR_CRASHED;
+      return -1;
+    }
+    return info->index_cond_func(info->index_cond_func_arg);
+  }
+  return 1;
+}
+
+
 /*
   Retrieve auto_increment info
 
diff --git a/storage/maria/ma_rkey.c b/storage/maria/ma_rkey.c
index 9e529aca666..32a0e141b1d 100644
--- a/storage/maria/ma_rkey.c
+++ b/storage/maria/ma_rkey.c
@@ -34,6 +34,7 @@ int maria_rkey(MARIA_HA *info, uchar *buf, int inx, const uchar *key_data,
   HA_KEYSEG *last_used_keyseg;
   uint32 nextflag;
   MARIA_KEY key;
+  int icp_res= 1;
   DBUG_ENTER("maria_rkey");
   DBUG_PRINT("enter", ("base: 0x%lx  buf: 0x%lx  inx: %d  search_flag: %d",
                        (long) info, (long) buf, inx, search_flag));
@@ -106,9 +107,13 @@ int maria_rkey(MARIA_HA *info, uchar *buf, int inx, const uchar *key_data,
         are inserted by other threads after we got our table lock
         ("concurrent inserts"). The record may not even be present yet.
         Keys are inserted into the index(es) before the record is
-        inserted into the data file. 
+        inserted into the data file.
+
+        If index condition is present, it must be either satisfied or 
+        not satisfied with an out-of-range condition.
       */
-      if ((*share->row_is_visible)(info))
+      if ((*share->row_is_visible)(info) && 
+          ((icp_res= ma_check_index_cond(info, inx, buf)) != 0))
         break;
 
       /* The key references a concurrently inserted record. */
@@ -120,7 +125,7 @@ int maria_rkey(MARIA_HA *info, uchar *buf, int inx, const uchar *key_data,
         info->cur_row.lastpos= HA_OFFSET_ERROR;
         break;
       }
-
+      
       do
       {
         uint not_used[2];
@@ -151,18 +156,25 @@ int maria_rkey(MARIA_HA *info, uchar *buf, int inx, const uchar *key_data,
           break;
           /* purecov: end */
         }
-      } while (!(*share->row_is_visible)(info));
+
+      } while (!(*share->row_is_visible)(info) || 
+               ((icp_res= ma_check_index_cond(info, inx, buf)) == 0));
     }
   }
   if (share->lock_key_trees)
     rw_unlock(&keyinfo->root_lock);
 
-  if (info->cur_row.lastpos == HA_OFFSET_ERROR)
+  if (info->cur_row.lastpos == HA_OFFSET_ERROR || (icp_res != 1))
   {
+    if (icp_res == 2)
+    {
+      info->cur_row.lastpos= HA_OFFSET_ERROR;
+      my_errno= HA_ERR_KEY_NOT_FOUND;
+    }
     fast_ma_writeinfo(info);
     goto err;
   }
-
+  
   /* Calculate length of the found key;  Used by maria_rnext_same */
   if ((keyinfo->flag & HA_VAR_LENGTH_KEY))
     info->last_rkey_length= _ma_keylength_part(keyinfo, info->lastkey_buff,
diff --git a/storage/maria/ma_rnext.c b/storage/maria/ma_rnext.c
index be960eccfe0..bdba5ff3a17 100644
--- a/storage/maria/ma_rnext.c
+++ b/storage/maria/ma_rnext.c
@@ -30,6 +30,7 @@ int maria_rnext(MARIA_HA *info, uchar *buf, int inx)
   uint flag;
   MARIA_SHARE *share= info->s;
   MARIA_KEYDEF *keyinfo;
+  int icp_res= 1;
   DBUG_ENTER("maria_rnext");
 
   if ((inx = _ma_check_index(info,inx)) < 0)
@@ -90,7 +91,8 @@ int maria_rnext(MARIA_HA *info, uchar *buf, int inx)
 
   if (!error)
   {
-    while (!(*share->row_is_visible)(info))
+    while (!(*share->row_is_visible)(info) ||
+           ((icp_res= ma_check_index_cond(info, inx, buf)) == 0))
     {
       /* Skip rows inserted by other threads since we got a lock */
       if  ((error= _ma_search_next(info, &info->last_key,
@@ -105,8 +107,11 @@ int maria_rnext(MARIA_HA *info, uchar *buf, int inx)
 	/* Don't clear if database-changed */
   info->update&= (HA_STATE_CHANGED | HA_STATE_ROW_CHANGED);
   info->update|= HA_STATE_NEXT_FOUND;
+  
+  if (icp_res == 2)
+    my_errno=HA_ERR_END_OF_FILE; /* got beyond the end of scanned range */
 
-  if (error)
+  if (error || icp_res != 1)
   {
     if (my_errno == HA_ERR_KEY_NOT_FOUND)
       my_errno=HA_ERR_END_OF_FILE;
diff --git a/storage/maria/ma_rnext_same.c b/storage/maria/ma_rnext_same.c
index cbd81d20816..597d97fc78a 100644
--- a/storage/maria/ma_rnext_same.c
+++ b/storage/maria/ma_rnext_same.c
@@ -30,6 +30,7 @@ int maria_rnext_same(MARIA_HA *info, uchar *buf)
   int error;
   uint inx,not_used[2];
   MARIA_KEYDEF *keyinfo;
+  int icp_res= 1;
   DBUG_ENTER("maria_rnext_same");
 
   if ((int) (inx= info->lastinx) < 0 ||
@@ -80,7 +81,8 @@ int maria_rnext_same(MARIA_HA *info, uchar *buf)
           break;
         }
         /* Skip rows that are inserted by other threads since we got a lock */
-        if ((info->s->row_is_visible)(info))
+        if ((info->s->row_is_visible)(info) ||
+            ((icp_res= ma_check_index_cond(info, inx, buf)) != 0))
           break;
       }
   }
@@ -90,7 +92,10 @@ int maria_rnext_same(MARIA_HA *info, uchar *buf)
   info->update&= (HA_STATE_CHANGED | HA_STATE_ROW_CHANGED);
   info->update|= HA_STATE_NEXT_FOUND | HA_STATE_RNEXT_SAME;
 
-  if (error)
+  if (icp_res == 2)
+    my_errno=HA_ERR_END_OF_FILE; /* got beyond the end of scanned range */
+
+  if (error || icp_res != 1)
   {
     if (my_errno == HA_ERR_KEY_NOT_FOUND)
       my_errno=HA_ERR_END_OF_FILE;
diff --git a/storage/maria/maria_def.h b/storage/maria/maria_def.h
index ada4ae3f426..cf36b76eaba 100644
--- a/storage/maria/maria_def.h
+++ b/storage/maria/maria_def.h
@@ -477,6 +477,8 @@ typedef struct st_maria_block_scan
   MARIA_RECORD_POS row_base_page;
 } MARIA_BLOCK_SCAN;
 
+/*psergey-todo: do really need to have copies of this all over the place?*/
+typedef my_bool (*index_cond_func_t)(void *param);
 
 struct st_maria_handler
 {
@@ -577,6 +579,9 @@ struct st_maria_handler
   uchar *maria_rtree_recursion_state;	/* For RTREE */
   uchar length_buff[5];			/* temp buff to store blob lengths */
   int maria_rtree_recursion_depth;
+
+  index_cond_func_t index_cond_func;   /* Index condition function */
+  void *index_cond_func_arg;           /* parameter for the func */
 };
 
 /* Some defines used by maria-functions */
@@ -1236,3 +1241,6 @@ extern my_bool maria_flush_log_for_page_none(uchar *page,
                                              uchar *data_ptr);
 void maria_concurrent_inserts(MARIA_HA *info, my_bool concurrent_insert);
 extern PAGECACHE *maria_log_pagecache;
+extern void ma_set_index_cond_func(MARIA_HA *info, index_cond_func_t func,
+                                   void *func_arg);
+int ma_check_index_cond(register MARIA_HA *info, uint keynr, uchar *record);
diff --git a/storage/myisam/ha_myisam.cc b/storage/myisam/ha_myisam.cc
index d92a576a5d7..7e319fc80f6 100644
--- a/storage/myisam/ha_myisam.cc
+++ b/storage/myisam/ha_myisam.cc
@@ -1676,6 +1676,48 @@ int ha_myisam::delete_row(const uchar *buf)
   return mi_delete(file,buf);
 }
 
+
+C_MODE_START
+
+ICP_RESULT index_cond_func_myisam(void *arg)
+{
+  ha_myisam *h= (ha_myisam*)arg;
+  if (h->end_range)
+  {
+    if (h->compare_key2(h->end_range) > 0)
+      return ICP_OUT_OF_RANGE; /* caller should return HA_ERR_END_OF_FILE already */
+  }
+  return (ICP_RESULT) test(h->pushed_idx_cond->val_int());
+}
+
+C_MODE_END
+
+
+int ha_myisam::index_init(uint idx, bool sorted)
+{ 
+  active_index=idx;
+  if (pushed_idx_cond_keyno == idx)
+    mi_set_index_cond_func(file, index_cond_func_myisam, this);
+  return 0; 
+}
+
+
+int ha_myisam::index_end()
+{
+  active_index=MAX_KEY;
+  //pushed_idx_cond_keyno= MAX_KEY;
+  mi_set_index_cond_func(file, NULL, 0);
+  in_range_check_pushed_down= FALSE;
+  ds_mrr.dsmrr_close();
+  return 0; 
+}
+
+int ha_myisam::rnd_end()
+{
+  ds_mrr.dsmrr_close();
+  return 0;
+}
+
 int ha_myisam::index_read_map(uchar *buf, const uchar *key,
                               key_part_map keypart_map,
                               enum ha_rkey_function find_flag)
@@ -1878,8 +1920,13 @@ int ha_myisam::extra(enum ha_extra_function operation)
   return mi_extra(file, operation, 0);
 }
 
+
 int ha_myisam::reset(void)
 {
+  pushed_idx_cond= NULL;
+  pushed_idx_cond_keyno= MAX_KEY;
+  mi_set_index_cond_func(file, NULL, 0);
+  ds_mrr.dsmrr_close();
   return mi_reset(file);
 }
 
@@ -2164,6 +2211,62 @@ static int myisam_init(void *p)
   return 0;
 }
 
+/****************************************************************************
+ * MyISAM MRR implementation: use DS-MRR
+ ***************************************************************************/
+
+int ha_myisam::multi_range_read_init(RANGE_SEQ_IF *seq, void *seq_init_param,
+                                     uint n_ranges, uint mode, 
+                                     HANDLER_BUFFER *buf)
+{
+  return ds_mrr.dsmrr_init(this, seq, seq_init_param, n_ranges, mode, buf);
+}
+
+int ha_myisam::multi_range_read_next(char **range_info)
+{
+  return ds_mrr.dsmrr_next(range_info);
+}
+
+ha_rows ha_myisam::multi_range_read_info_const(uint keyno, RANGE_SEQ_IF *seq,
+                                               void *seq_init_param, 
+                                               uint n_ranges, uint *bufsz,
+                                               uint *flags, COST_VECT *cost)
+{
+  /*
+    This call is here because there is no location where this->table would
+    already be known.
+    TODO: consider moving it into some per-query initialization call.
+  */
+  ds_mrr.init(this, table);
+  return ds_mrr.dsmrr_info_const(keyno, seq, seq_init_param, n_ranges, bufsz,
+                                 flags, cost);
+}
+
+ha_rows ha_myisam::multi_range_read_info(uint keyno, uint n_ranges, uint keys,
+                                         uint *bufsz, uint *flags,
+                                         COST_VECT *cost)
+{
+  ds_mrr.init(this, table);
+  return ds_mrr.dsmrr_info(keyno, n_ranges, keys, bufsz, flags, cost);
+}
+
+/* MyISAM MRR implementation ends */
+
+
+/* Index condition pushdown implementation*/
+
+
+Item *ha_myisam::idx_cond_push(uint keyno_arg, Item* idx_cond_arg)
+{
+  pushed_idx_cond_keyno= keyno_arg;
+  pushed_idx_cond= idx_cond_arg;
+  in_range_check_pushed_down= TRUE;
+  if (active_index == pushed_idx_cond_keyno)
+    mi_set_index_cond_func(file, index_cond_func_myisam, this);
+  return NULL;
+}
+
+
 struct st_mysql_storage_engine myisam_storage_engine=
 { MYSQL_HANDLERTON_INTERFACE_VERSION };
 
diff --git a/storage/myisam/ha_myisam.h b/storage/myisam/ha_myisam.h
index 7a1595573d4..76db0e89536 100644
--- a/storage/myisam/ha_myisam.h
+++ b/storage/myisam/ha_myisam.h
@@ -34,6 +34,10 @@ extern ulong myisam_sort_buffer_size;
 extern TYPELIB myisam_recover_typelib;
 extern ulong myisam_recover_options;
 
+C_MODE_START
+ICP_RESULT index_cond_func_myisam(void *arg);
+C_MODE_END
+
 class ha_myisam: public handler
 {
   MI_INFO *file;
@@ -50,11 +54,15 @@ class ha_myisam: public handler
   const char *index_type(uint key_number);
   const char **bas_ext() const;
   ulonglong table_flags() const { return int_table_flags; }
+  int index_init(uint idx, bool sorted);
+  int index_end();
+  int rnd_end();
+
   ulong index_flags(uint inx, uint part, bool all_parts) const
   {
     return ((table_share->key_info[inx].algorithm == HA_KEY_ALG_FULLTEXT) ?
             0 : HA_READ_NEXT | HA_READ_PREV | HA_READ_RANGE |
-            HA_READ_ORDER | HA_KEYREAD_ONLY);
+            HA_READ_ORDER | HA_KEYREAD_ONLY | HA_DO_INDEX_COND_PUSHDOWN);
   }
   uint max_supported_keys()          const { return MI_MAX_KEY; }
   uint max_supported_key_length()    const { return HA_MAX_KEY_LENGTH; }
@@ -149,4 +157,23 @@ class ha_myisam: public handler
   {
     return file;
   }
+public:
+  /**
+   * Multi Range Read interface
+   */
+  int multi_range_read_init(RANGE_SEQ_IF *seq, void *seq_init_param,
+                            uint n_ranges, uint mode, HANDLER_BUFFER *buf);
+  int multi_range_read_next(char **range_info);
+  ha_rows multi_range_read_info_const(uint keyno, RANGE_SEQ_IF *seq,
+                                      void *seq_init_param, 
+                                      uint n_ranges, uint *bufsz,
+                                      uint *flags, COST_VECT *cost);
+  ha_rows multi_range_read_info(uint keyno, uint n_ranges, uint keys,
+                                uint *bufsz, uint *flags, COST_VECT *cost);
+  
+  /* Index condition pushdown implementation */
+  Item *idx_cond_push(uint keyno, Item* idx_cond);
+private:
+  DsMrr_impl ds_mrr;
+  friend ICP_RESULT index_cond_func_myisam(void *arg);
 };
diff --git a/storage/myisam/mi_extra.c b/storage/myisam/mi_extra.c
index 9c5dade28ab..7bb79108357 100644
--- a/storage/myisam/mi_extra.c
+++ b/storage/myisam/mi_extra.c
@@ -403,6 +403,12 @@ int mi_extra(MI_INFO *info, enum ha_extra_function function, void *extra_arg)
   DBUG_RETURN(error);
 } /* mi_extra */
 
+void mi_set_index_cond_func(MI_INFO *info, index_cond_func_t func,
+                            void *func_arg)
+{
+  info->index_cond_func= func;
+  info->index_cond_func_arg= func_arg;
+}
 
 /*
     Start/Stop Inserting Duplicates Into a Table, WL#1648.
diff --git a/storage/myisam/mi_key.c b/storage/myisam/mi_key.c
index 94f3f34ec58..aab5797a03b 100644
--- a/storage/myisam/mi_key.c
+++ b/storage/myisam/mi_key.c
@@ -31,7 +31,8 @@
               set_if_smaller(char_length,length);                           \
             } while(0)
 
-static int _mi_put_key_in_record(MI_INFO *info,uint keynr,uchar *record);
+static int _mi_put_key_in_record(MI_INFO *info,uint keynr,
+                                 my_bool unpack_blobs, uchar *record);
 
 /*
   Make a intern key from a record
@@ -312,6 +313,9 @@ uint _mi_pack_key(register MI_INFO *info, uint keynr, uchar *key, uchar *old,
     _mi_put_key_in_record()
     info		MyISAM handler
     keynr		Key number that was used
+    unpack_blobs        TRUE  <=> Unpack blob columns
+                        FALSE <=> Skip them. This is used by index condition 
+                                  pushdown check function
     record 		Store key here
 
     Last read key is in info->lastkey
@@ -325,7 +329,7 @@ uint _mi_pack_key(register MI_INFO *info, uint keynr, uchar *key, uchar *old,
 */
 
 static int _mi_put_key_in_record(register MI_INFO *info, uint keynr,
-				 uchar *record)
+                                 my_bool unpack_blobs, uchar *record)
 {
   reg2 uchar *key;
   uchar *pos,*key_end;
@@ -418,16 +422,19 @@ static int _mi_put_key_in_record(register MI_INFO *info, uint keynr,
       if (length > keyseg->length || key+length > key_end)
 	goto err;
 #endif
-      memcpy(record+keyseg->start+keyseg->bit_start,
-	     (char*) &blob_ptr,sizeof(char*));
-      memcpy(blob_ptr,key,length);
-      blob_ptr+=length;
+      if (unpack_blobs)
+      {
+        memcpy(record+keyseg->start+keyseg->bit_start,
+               (char*) &blob_ptr,sizeof(char*));
+        memcpy(blob_ptr,key,length);
+        blob_ptr+=length;
 
-      /* The above changed info->lastkey2. Inform mi_rnext_same(). */
-      info->update&= ~HA_STATE_RNEXT_SAME;
+        /* The above changed info->lastkey2. Inform mi_rnext_same(). */
+        info->update&= ~HA_STATE_RNEXT_SAME;
 
-      _mi_store_blob_length(record+keyseg->start,
-			    (uint) keyseg->bit_start,length);
+        _mi_store_blob_length(record+keyseg->start,
+                              (uint) keyseg->bit_start,length);
+      }
       key+=length;
     }
     else if (keyseg->flag & HA_SWAP_KEY)
@@ -471,7 +478,7 @@ int _mi_read_key_record(MI_INFO *info, my_off_t filepos, uchar *buf)
   {
     if (info->lastinx >= 0)
     {				/* Read only key */
-      if (_mi_put_key_in_record(info,(uint) info->lastinx,buf))
+      if (_mi_put_key_in_record(info,(uint) info->lastinx, TRUE, buf))
       {
         mi_print_error(info->s, HA_ERR_CRASHED);
 	my_errno=HA_ERR_CRASHED;
@@ -487,6 +494,34 @@ int _mi_read_key_record(MI_INFO *info, my_off_t filepos, uchar *buf)
 
 
 /*
+  Save current key tuple to record and call index condition check function
+
+  SYNOPSIS
+    mi_check_index_cond()
+      info    MyISAM handler
+      keynr   Index we're running a scan on
+      record  Record buffer to use (it is assumed that index check function 
+              will look for column values there)
+
+  RETURN
+    -1  Error 
+    0   Index condition is not satisfied, continue scanning
+    1   Index condition is satisfied
+    2   Index condition is not satisfied, end the scan. 
+*/
+
+int mi_check_index_cond(register MI_INFO *info, uint keynr, uchar *record)
+{
+  if (_mi_put_key_in_record(info, keynr, FALSE, record))
+  {
+    mi_print_error(info->s, HA_ERR_CRASHED);
+    my_errno=HA_ERR_CRASHED;
+    return -1;
+  }
+  return info->index_cond_func(info->index_cond_func_arg);
+}
+
+/*
   Retrieve auto_increment info
 
   SYNOPSIS
diff --git a/storage/myisam/mi_rkey.c b/storage/myisam/mi_rkey.c
index f20b0366683..47e2ce51a9e 100644
--- a/storage/myisam/mi_rkey.c
+++ b/storage/myisam/mi_rkey.c
@@ -29,6 +29,7 @@ int mi_rkey(MI_INFO *info, uchar *buf, int inx, const uchar *key,
   MI_KEYDEF *keyinfo;
   HA_KEYSEG *last_used_keyseg;
   uint pack_key_length, use_key_length, nextflag;
+  int res= 0;
   DBUG_ENTER("mi_rkey");
   DBUG_PRINT("enter", ("base: 0x%lx  buf: 0x%lx  inx: %d  search_flag: %d",
                        (long) info, (long) buf, inx, search_flag));
@@ -105,55 +106,62 @@ int mi_rkey(MI_INFO *info, uchar *buf, int inx, const uchar *key,
         saved the current data_file_length. Concurrent inserts always go
         to the end of the file. So we can test if the found key
         references a new record.
+
+        If we are searching for a partial key (or using >, >=, < or <=) and
+        the data is outside of the data file, we need to continue searching
+        for the first key inside the data file.
+
+        We do also continue searching if an index condition check function
+        is available.
       */
-      if (info->lastpos >= info->state->data_file_length)
+      while ((info->lastpos >= info->state->data_file_length &&
+              (search_flag != HA_READ_KEY_EXACT ||
+              last_used_keyseg != keyinfo->seg + keyinfo->keysegs)) ||
+             (info->index_cond_func && 
+              !(res= mi_check_index_cond(info, inx, buf))))
       {
-        /* The key references a concurrently inserted record. */
+        uint not_used[2];
+        /*
+          Skip rows that are inserted by other threads since we got a lock
+          Note that this can only happen if we are not searching after an
+          full length exact key, because the keys are sorted
+          according to position
+        */
+        if  (_mi_search_next(info, keyinfo, info->lastkey,
+                             info->lastkey_length,
+                             myisam_readnext_vec[search_flag],
+                             info->s->state.key_root[inx]))
+          break;
+        /*
+          Check that the found key does still match the search.
+          _mi_search_next() delivers the next key regardless of its
+          value.
+        */
         if (search_flag == HA_READ_KEY_EXACT &&
-            last_used_keyseg == keyinfo->seg + keyinfo->keysegs)
+            ha_key_cmp(keyinfo->seg, key_buff, info->lastkey, use_key_length,
+                       SEARCH_FIND, not_used))
         {
-          /* Simply ignore the key if it matches exactly. (Bug #29838) */
           my_errno= HA_ERR_KEY_NOT_FOUND;
           info->lastpos= HA_OFFSET_ERROR;
+          break;
         }
-        else
-        {
-          /*
-            If searching for a partial key (or using >, >=, < or <=) and
-            the data is outside of the data file, we need to continue
-            searching for the first key inside the data file.
-          */
-          do
-          {
-            uint not_used[2];
-            /*
-              Skip rows that are inserted by other threads since we got
-              a lock. Note that this can only happen if we are not
-              searching after a full length exact key, because the keys
-              are sorted according to position.
-            */
-            if  (_mi_search_next(info, keyinfo, info->lastkey,
-                                 info->lastkey_length,
-                                 myisam_readnext_vec[search_flag],
-                                 info->s->state.key_root[inx]))
-              break; /* purecov: inspected */
-            /*
-              Check that the found key does still match the search.
-              _mi_search_next() delivers the next key regardless of its
-              value.
-            */
-            if (search_flag == HA_READ_KEY_EXACT &&
-                ha_key_cmp(keyinfo->seg, key_buff, info->lastkey,
-                           use_key_length, SEARCH_FIND, not_used))
-            {
-              /* purecov: begin inspected */
-              my_errno= HA_ERR_KEY_NOT_FOUND;
-              info->lastpos= HA_OFFSET_ERROR;
-              break;
-              /* purecov: end */
-            }
-          } while (info->lastpos >= info->state->data_file_length);
-        }
+      }
+      if (res == 2)
+      {
+        info->lastpos= HA_OFFSET_ERROR;
+        if (share->concurrent_insert)
+          rw_unlock(&share->key_root_lock[inx]);
+        DBUG_RETURN((my_errno= HA_ERR_KEY_NOT_FOUND));
+      }
+      /*
+        Error if no row found within the data file. (Bug #29838)
+        Do not overwrite my_errno if already at HA_OFFSET_ERROR.
+      */
+      if (info->lastpos != HA_OFFSET_ERROR &&
+          info->lastpos >= info->state->data_file_length)
+      {
+        info->lastpos= HA_OFFSET_ERROR;
+        my_errno= HA_ERR_KEY_NOT_FOUND;
       }
     }
   }
diff --git a/storage/myisam/mi_rnext.c b/storage/myisam/mi_rnext.c
index 7ce66d41e0f..ad5ec60dc98 100644
--- a/storage/myisam/mi_rnext.c
+++ b/storage/myisam/mi_rnext.c
@@ -28,6 +28,7 @@ int mi_rnext(MI_INFO *info, uchar *buf, int inx)
 {
   int error,changed;
   uint flag;
+  int res= 0;
   DBUG_ENTER("mi_rnext");
 
   if ((inx = _mi_check_index(info,inx)) < 0)
@@ -81,23 +82,36 @@ int mi_rnext(MI_INFO *info, uchar *buf, int inx)
     }
   }
 
-  if (info->s->concurrent_insert)
+  if (!error)
   {
-    if (!error)
+    while ((info->s->concurrent_insert &&
+            info->lastpos >= info->state->data_file_length) ||
+           (info->index_cond_func &&
+           !(res= mi_check_index_cond(info, inx, buf))))
     {
-      while (info->lastpos >= info->state->data_file_length)
-      {
-	/* Skip rows inserted by other threads since we got a lock */
-	if  ((error=_mi_search_next(info,info->s->keyinfo+inx,
-				    info->lastkey,
-				    info->lastkey_length,
-				    SEARCH_BIGGER,
-				    info->s->state.key_root[inx])))
-	  break;
-      }
+      /* 
+         Skip rows that are either inserted by other threads since
+         we got a lock or do not match pushed index conditions
+      */
+      if  ((error=_mi_search_next(info,info->s->keyinfo+inx,
+                                  info->lastkey,
+                                  info->lastkey_length,
+                                  SEARCH_BIGGER,
+                                  info->s->state.key_root[inx])))
+        break;
+    }
+    if (!error && res == 2)
+    {
+      if (info->s->concurrent_insert)
+        rw_unlock(&info->s->key_root_lock[inx]);
+      info->lastpos= HA_OFFSET_ERROR;
+      DBUG_RETURN(my_errno= HA_ERR_END_OF_FILE);
     }
-    rw_unlock(&info->s->key_root_lock[inx]);
   }
+  
+  if (info->s->concurrent_insert)
+    rw_unlock(&info->s->key_root_lock[inx]);
+
 	/* Don't clear if database-changed */
   info->update&= (HA_STATE_CHANGED | HA_STATE_ROW_CHANGED);
   info->update|= HA_STATE_NEXT_FOUND;
diff --git a/storage/myisam/mi_rnext_same.c b/storage/myisam/mi_rnext_same.c
index 1892fe3e1e0..76775dec8ac 100644
--- a/storage/myisam/mi_rnext_same.c
+++ b/storage/myisam/mi_rnext_same.c
@@ -76,7 +76,8 @@ int mi_rnext_same(MI_INFO *info, uchar *buf)
           break;
         }
         /* Skip rows that are inserted by other threads since we got a lock */
-        if (info->lastpos < info->state->data_file_length)
+        if (info->lastpos < info->state->data_file_length && 
+            (!info->index_cond_func || mi_check_index_cond(info, inx, buf)))
           break;
       }
   }
diff --git a/storage/myisam/mi_rprev.c b/storage/myisam/mi_rprev.c
index d1407012590..93bd224bd56 100644
--- a/storage/myisam/mi_rprev.c
+++ b/storage/myisam/mi_rprev.c
@@ -51,22 +51,36 @@ int mi_rprev(MI_INFO *info, uchar *buf, int inx)
     error=_mi_search(info,share->keyinfo+inx,info->lastkey,
 		     USE_WHOLE_KEY, flag, share->state.key_root[inx]);
 
-  if (share->concurrent_insert)
+  if (!error)
   {
-    if (!error)
+    int res= 0;
+    while ((share->concurrent_insert && 
+            info->lastpos >= info->state->data_file_length) ||
+           (info->index_cond_func &&
+            !(res= mi_check_index_cond(info, inx, buf))))
     {
-      while (info->lastpos >= info->state->data_file_length)
-      {
-	/* Skip rows that are inserted by other threads since we got a lock */
-	if  ((error=_mi_search_next(info,share->keyinfo+inx,info->lastkey,
-				    info->lastkey_length,
-				    SEARCH_SMALLER,
-				    share->state.key_root[inx])))
-	  break;
-      }
+      /* 
+         Skip rows that are either inserted by other threads since
+         we got a lock or do not match pushed index conditions
+      */
+      if  ((error=_mi_search_next(info,share->keyinfo+inx,info->lastkey,
+                                  info->lastkey_length,
+                                  SEARCH_SMALLER,
+                                  share->state.key_root[inx])))
+        break;
+    }
+    if (!error && res == 2) 
+    {
+      if (share->concurrent_insert)
+        rw_unlock(&share->key_root_lock[inx]);
+      info->lastpos= HA_OFFSET_ERROR;
+      DBUG_RETURN(my_errno= HA_ERR_END_OF_FILE);
     }
-    rw_unlock(&share->key_root_lock[inx]);
   }
+
+  if (share->concurrent_insert)
+    rw_unlock(&share->key_root_lock[inx]);
+
   info->update&= (HA_STATE_CHANGED | HA_STATE_ROW_CHANGED);
   info->update|= HA_STATE_PREV_FOUND;
   if (error)
diff --git a/storage/myisam/myisamdef.h b/storage/myisam/myisamdef.h
index c2841c49199..2dcd008da58 100644
--- a/storage/myisam/myisamdef.h
+++ b/storage/myisam/myisamdef.h
@@ -232,6 +232,7 @@ typedef struct st_mi_isam_share
   rw_lock_t mmap_lock;
 } MYISAM_SHARE;
 
+typedef ICP_RESULT (*index_cond_func_t)(void *param);
 
 struct st_myisam_info
 {
@@ -300,6 +301,8 @@ struct st_myisam_info
   /* If info->buff has to be reread for rnext */
   my_bool buff_used;
   my_bool once_flags;                   /* For MYISAMMRG */
+  index_cond_func_t index_cond_func;   /* Index condition function */
+  void *index_cond_func_arg;           /* parameter for the func */
 #ifdef __WIN__
   my_bool owned_by_merge;                       /* This MyISAM table is part of a merge union */
 #endif
@@ -724,6 +727,7 @@ void mi_setup_functions(register MYISAM_SHARE *share);
 my_bool mi_dynmap_file(MI_INFO *info, my_off_t size);
 void mi_remap_file(MI_INFO *info, my_off_t size);
 
+int mi_check_index_cond(register MI_INFO *info, uint keynr, uchar *record);
     /* Functions needed by mi_check */
 volatile int *killed_ptr(HA_CHECK *param);
 void mi_check_print_error _VARARGS((HA_CHECK *param, const char *fmt, ...));
@@ -732,6 +736,8 @@ void mi_check_print_info _VARARGS((HA_CHECK *param, const char *fmt, ...));
 #ifdef THREAD
 pthread_handler_t thr_find_all_keys(void *arg);
 #endif
+extern void mi_set_index_cond_func(MI_INFO *info, index_cond_func_t func,
+                                   void *func_arg);
 int flush_blocks(HA_CHECK *param, KEY_CACHE *key_cache, File file);
 #ifdef __cplusplus
 }
diff --git a/storage/xtradb/handler/ha_innodb.cc b/storage/xtradb/handler/ha_innodb.cc
index 1cd4000846a..aa6e6a27a69 100644
--- a/storage/xtradb/handler/ha_innodb.cc
+++ b/storage/xtradb/handler/ha_innodb.cc
@@ -113,6 +113,12 @@ static pthread_cond_t commit_cond;
 static pthread_mutex_t commit_cond_m;
 static bool innodb_inited = 0;
 
+C_MODE_START
+static uint index_cond_func_innodb(void *arg);
+C_MODE_END
+
+
+
 #define INSIDE_HA_INNOBASE_CC
 
 /* In the Windows plugin, the return value of current_thd is
@@ -3031,7 +3037,7 @@ ulong
 ha_innobase::index_flags(uint, uint, bool) const
 {
 	return(HA_READ_NEXT | HA_READ_PREV | HA_READ_ORDER
-	       | HA_READ_RANGE | HA_KEYREAD_ONLY);
+	       | HA_READ_RANGE | HA_KEYREAD_ONLY | HA_DO_INDEX_COND_PUSHDOWN);
 }
 
 UNIV_INTERN
@@ -3989,6 +3995,7 @@ build_template(
 					only if templ_type is
 					ROW_MYSQL_REC_FIELDS */
 	TABLE*		table,		/* in: MySQL table */
+        ha_innobase*    file,           /* in: ha_innobase handler */
 	uint		templ_type)	/* in: ROW_MYSQL_WHOLE_ROW or
 					ROW_MYSQL_REC_FIELDS */
 {
@@ -4000,10 +4007,12 @@ build_template(
 	ulint		n_requested_fields	= 0;
 	ibool		fetch_all_in_key	= FALSE;
 	ibool		fetch_primary_key_cols	= FALSE;
-	ulint		i, sql_idx, innodb_idx=0;
+	ulint		sql_idx, innodb_idx=0;
 	/* byte offset of the end of last requested column */
 	ulint		mysql_prefix_len	= 0;
-
+        ibool           do_idx_cond_push= FALSE;
+	ibool           need_second_pass= FALSE;
+        
 	if (prebuilt->select_lock_type == LOCK_X) {
 		/* We always retrieve the whole clustered index record if we
 		use exclusive row level locks, for example, if the read is
@@ -4073,6 +4082,16 @@ build_template(
 
 	prebuilt->templ_contains_blob = FALSE;
 
+        
+        /*
+          Setup index condition pushdown (note: we don't need to check if
+          this is a scan on primary key as that is checked in idx_cond_push)
+        */
+        if (file->active_index == file->pushed_idx_cond_keyno && 
+            file->active_index != MAX_KEY && 
+            templ_type == ROW_MYSQL_REC_FIELDS)
+          do_idx_cond_push= need_second_pass= TRUE;
+
 	/* Note that in InnoDB, i is the column number. MySQL calls columns
 	'fields'. */
 	for (sql_idx = 0; sql_idx < n_fields; sql_idx++) {
@@ -4086,6 +4105,8 @@ build_template(
 			and which we can skip. */
 			register const ibool	index_contains_field =
 				dict_index_contains_col_or_prefix(index, innodb_idx);
+                        register const ibool    index_covers_field = 
+                                field->part_of_key.is_set(file->active_index);
 
 			if (!index_contains_field && prebuilt->read_just_key) {
 				/* If this is a 'key read', we do not need
@@ -4118,8 +4139,12 @@ build_template(
 			/* This field is not needed in the query, skip it */
 
 			goto skip_field;
-		}
 include_field:
+			if (do_idx_cond_push && 
+                            ((need_second_pass && !index_covers_field) || 
+                             (!need_second_pass && index_covers_field)))
+			  goto skip_field;
+		}
 		n_requested_fields++;
 
 		templ->col_no = innodb_idx;
@@ -4173,6 +4198,13 @@ include_field:
 			prebuilt->templ_contains_blob = TRUE;
 		}
 skip_field:
+		if (need_second_pass && (sql_idx+1 == n_fields))
+		{
+                  prebuilt->n_index_fields= n_requested_fields;
+		  need_second_pass= FALSE;
+		  sql_idx= (~(ulint)0); /* to start from 0 */
+		  innodb_idx= (~(ulint)0); /* to start from 0 */ ///psergey-merge-merge-last-change
+		}
                 if (field->stored_in_db) {
                     innodb_idx++;
                 }
@@ -4181,12 +4213,23 @@ skip_field:
 	prebuilt->n_template = n_requested_fields;
 	prebuilt->mysql_prefix_len = mysql_prefix_len;
 
+        if (do_idx_cond_push)
+        {
+          prebuilt->idx_cond_func= index_cond_func_innodb;
+          prebuilt->idx_cond_func_arg= file;
+        }
+        else
+        {
+          prebuilt->idx_cond_func= NULL;
+          prebuilt->n_index_fields= n_requested_fields;
+        }
+
 	if (index != clust_index && prebuilt->need_to_access_clustered) {
 		/* Change rec_field_no's to correspond to the clustered index
 		record */
-		for (i = 0; i < n_requested_fields; i++) {
+		for (ulint i = do_idx_cond_push? prebuilt->n_index_fields : 0; 
+                     i < n_requested_fields; i++) {
 			templ = prebuilt->mysql_template + i;
-
 			templ->rec_field_no = dict_col_get_clust_pos(
 				&index->table->cols[templ->col_no],
 				clust_index);
@@ -4505,7 +4548,7 @@ no_commit:
 		/* Build the template used in converting quickly between
 		the two database formats */
 
-		build_template(prebuilt, NULL, table, ROW_MYSQL_WHOLE_ROW);
+		build_template(prebuilt, NULL, table, this, ROW_MYSQL_WHOLE_ROW);
 	}
 
 	innodb_srv_conc_enter_innodb(prebuilt->trx);
@@ -5038,6 +5081,8 @@ ha_innobase::index_end(void)
 	int	error	= 0;
 	DBUG_ENTER("index_end");
 	active_index=MAX_KEY;
+	in_range_check_pushed_down= FALSE;
+	ds_mrr.dsmrr_close();
 	DBUG_RETURN(error);
 }
 
@@ -5187,7 +5232,8 @@ ha_innobase::index_read(
 	necessarily prebuilt->index, but can also be the clustered index */
 
 	if (prebuilt->sql_stat_start) {
-		build_template(prebuilt, user_thd, table, ROW_MYSQL_REC_FIELDS);
+		build_template(prebuilt, user_thd, table, this,
+                               ROW_MYSQL_REC_FIELDS);
 	}
 
 	if (key_ptr) {
@@ -5359,7 +5405,7 @@ ha_innobase::change_active_index(
 	the flag ROW_MYSQL_WHOLE_ROW below, but that caused unnecessary
 	copying. Starting from MySQL-4.1 we use a more efficient flag here. */
 
-	build_template(prebuilt, user_thd, table, ROW_MYSQL_REC_FIELDS);
+	build_template(prebuilt, user_thd, table, this, ROW_MYSQL_REC_FIELDS);
 
 	DBUG_RETURN(0);
 }
@@ -7422,7 +7468,7 @@ ha_innobase::check(
 		/* Build the template; we will use a dummy template
 		in index scans done in checking */
 
-		build_template(prebuilt, NULL, table, ROW_MYSQL_WHOLE_ROW);
+		build_template(prebuilt, NULL, table, this, ROW_MYSQL_WHOLE_ROW);
 	}
 
 	ret = row_check_table_for_mysql(prebuilt);
@@ -7786,6 +7832,10 @@ ha_innobase::extra(
 			break;
 		case HA_EXTRA_RESET_STATE:
 			reset_template(prebuilt);
+                        /* Reset index condition pushdown state */
+                        pushed_idx_cond= FALSE;
+                        pushed_idx_cond_keyno= MAX_KEY;
+                        prebuilt->idx_cond_func= NULL;
 			break;
 		case HA_EXTRA_NO_KEYREAD:
 			prebuilt->read_just_key = 0;
@@ -7833,6 +7883,12 @@ ha_innobase::reset()
 
 	reset_template(prebuilt);
 
+	/* Reset index condition pushdown state */
+	pushed_idx_cond_keyno= MAX_KEY;
+	pushed_idx_cond= NULL;
+	ds_mrr.dsmrr_close();
+	prebuilt->idx_cond_func= NULL;
+
 	/* TODO: This should really be reset in reset_template() but for now
 	it's safer to do it explicitly here. */
 
@@ -10680,3 +10736,112 @@ test_innobase_convert_name()
 }
 
 #endif /* UNIV_COMPILE_TEST_FUNCS */
+
+
+/****************************************************************************
+ * DS-MRR implementation 
+ ***************************************************************************/
+
+/**
+ * Multi Range Read interface, DS-MRR calls
+ */
+
+int ha_innobase::multi_range_read_init(RANGE_SEQ_IF *seq, void *seq_init_param,
+                          uint n_ranges, uint mode, HANDLER_BUFFER *buf)
+{
+  return ds_mrr.dsmrr_init(this, seq, seq_init_param, n_ranges, mode, buf);
+}
+
+int ha_innobase::multi_range_read_next(char **range_info)
+{
+  return ds_mrr.dsmrr_next(range_info);
+}
+
+ha_rows ha_innobase::multi_range_read_info_const(uint keyno, RANGE_SEQ_IF *seq,
+                                                 void *seq_init_param,  
+                                                 uint n_ranges, uint *bufsz,
+                                                 uint *flags, 
+                                                 COST_VECT *cost)
+{
+  /* See comments in ha_myisam::multi_range_read_info_const */
+  ds_mrr.init(this, table);
+  //psergey-mrr-fix:
+  if (prebuilt->select_lock_type != LOCK_NONE)
+    *flags |= HA_MRR_USE_DEFAULT_IMPL;
+
+  uint orig_flags= *flags;
+
+  ha_rows res= ds_mrr.dsmrr_info_const(keyno, seq, seq_init_param, n_ranges,
+                                       bufsz, flags, cost);
+
+  bool disable_ds_mrr= true;
+  disable_ds_mrr= false;
+//  DBUG_EXECUTE_IF("optimizer_innodb_ds_mrr", disable_ds_mrr= false;);
+  if (!disable_ds_mrr)
+    return res;
+
+  /* Disable DS-MRR: enable MS-MRR only after critical bugs are fixed */
+  *bufsz= 0;
+  *flags = orig_flags | HA_MRR_USE_DEFAULT_IMPL;
+  return res;
+}
+
+ha_rows ha_innobase::multi_range_read_info(uint keyno, uint n_ranges, 
+                                           uint keys, uint *bufsz, 
+                                           uint *flags, COST_VECT *cost)
+{
+  ds_mrr.init(this, table);
+  uint orig_flags= *flags;
+
+  ha_rows res= ds_mrr.dsmrr_info(keyno, n_ranges, keys, bufsz, flags, cost);
+  bool disable_ds_mrr= false;
+ // DBUG_EXECUTE_IF("optimizer_innodb_ds_mrr", disable_ds_mrr= false;);
+  if (!disable_ds_mrr)
+    return res;
+
+  /* Disable DS-MRR: enable MS-MRR only after critical bugs are fixed */
+  *bufsz= 0;
+  *flags = orig_flags | HA_MRR_USE_DEFAULT_IMPL;
+  return res;
+}
+
+
+
+/**
+ * Index Condition Pushdown interface implementation
+ */
+
+C_MODE_START
+
+/*
+  Index condition check function to be called from within Innobase.
+  See note on ICP_RESULT for return values description.
+*/
+
+static uint index_cond_func_innodb(void *arg)
+{
+  ha_innobase *h= (ha_innobase*)arg;
+  if (h->end_range)
+  {
+    if (h->compare_key2(h->end_range) > 0)
+      return 2; /* caller should return HA_ERR_END_OF_FILE already */
+  }
+  return test(h->pushed_idx_cond->val_int());
+}
+
+C_MODE_END
+
+
+Item *ha_innobase::idx_cond_push(uint keyno_arg, Item* idx_cond_arg)
+{
+  //                              V :psergey-mrrr-merge: V
+  if (keyno_arg != primary_key && (prebuilt->select_lock_type == LOCK_NONE))
+  {
+    pushed_idx_cond_keyno= keyno_arg;
+    pushed_idx_cond= idx_cond_arg;
+    in_range_check_pushed_down= TRUE;
+    return NULL; /* Table handler will check the entire condition */
+  }
+  return idx_cond_arg; /* Table handler will not make any checks */
+}
+
diff --git a/storage/xtradb/handler/ha_innodb.h b/storage/xtradb/handler/ha_innodb.h
index cf58a3ed972..d5a50a74d6f 100644
--- a/storage/xtradb/handler/ha_innodb.h
+++ b/storage/xtradb/handler/ha_innodb.h
@@ -205,6 +205,22 @@ class ha_innobase: public handler
 	bool check_if_incompatible_data(HA_CREATE_INFO *info,
 					uint table_changes);
 	bool check_if_supported_virtual_columns(void) { return TRUE; }
+public:
+  /**
+   * Multi Range Read interface
+   */
+  int multi_range_read_init(RANGE_SEQ_IF *seq, void *seq_init_param,
+                            uint n_ranges, uint mode, HANDLER_BUFFER *buf);
+  int multi_range_read_next(char **range_info);
+  ha_rows multi_range_read_info_const(uint keyno, RANGE_SEQ_IF *seq,
+                                      void *seq_init_param, 
+                                      uint n_ranges, uint *bufsz,
+                                      uint *flags, COST_VECT *cost);
+  ha_rows multi_range_read_info(uint keyno, uint n_ranges, uint keys,
+                                uint *bufsz, uint *flags, COST_VECT *cost);
+  DsMrr_impl ds_mrr;
+
+  Item *idx_cond_push(uint keyno, Item* idx_cond);
 };
 
 /* Some accessor functions which the InnoDB plugin needs, but which
diff --git a/storage/xtradb/include/row0mysql.h b/storage/xtradb/include/row0mysql.h
index 8e42c316209..e1a73cab58d 100644
--- a/storage/xtradb/include/row0mysql.h
+++ b/storage/xtradb/include/row0mysql.h
@@ -564,6 +564,8 @@ struct mysql_row_templ_struct {
 #define ROW_PREBUILT_ALLOCATED	78540783
 #define ROW_PREBUILT_FREED	26423527
 
+typedef uint (*index_cond_func_t)(void *param);
+
 /* A struct for (sometimes lazily) prebuilt structures in an Innobase table
 handle used within MySQL; these are used to save CPU time. */
 
@@ -756,6 +758,14 @@ struct row_prebuilt_struct {
 	/*----------------------*/
 	UT_LIST_NODE_T(row_prebuilt_t)	prebuilts;
 					/* list node of table->prebuilts */
+	/*----------------------*/
+        index_cond_func_t idx_cond_func;/* Index Condition Pushdown function,
+                                        or NULL if there is none set */
+        void*           idx_cond_func_arg;/* ICP function  argument */
+        ulint           n_index_fields; /* Number of fields at the start of
+                                        mysql_template. Valid only when using
+                                        ICP. */
+	/*----------------------*/
 	ulint		magic_n2;	/* this should be the same as
 					magic_n */
 };
diff --git a/storage/xtradb/row/row0sel.c b/storage/xtradb/row/row0sel.c
index fb1523d3370..95df05c8ff0 100644
--- a/storage/xtradb/row/row0sel.c
+++ b/storage/xtradb/row/row0sel.c
@@ -2682,8 +2682,10 @@ row_sel_store_mysql_rec(
 					which was described in prebuilt's
 					template; must be protected by
 					a page latch */
-	const ulint*	offsets)	/* in: array returned by
+	const ulint*	offsets,	/* in: array returned by
 					rec_get_offsets() */
+	ulint		start_field_no,	/* in: start from this field */
+	ulint		end_field_no)	/* in: end at this field */
 {
 	mysql_row_templ_t*	templ;
 	mem_heap_t*		extern_field_heap	= NULL;
@@ -2701,7 +2703,7 @@ row_sel_store_mysql_rec(
 		prebuilt->blob_heap = NULL;
 	}
 
-	for (i = 0; i < prebuilt->n_template; i++) {
+	for (i = start_field_no; i < end_field_no /* prebuilt->n_template */ ; i++) {
 
 		templ = prebuilt->mysql_template + i;
 
@@ -3143,7 +3145,10 @@ row_sel_push_cache_row_for_mysql(
 	row_prebuilt_t*	prebuilt,	/* in: prebuilt struct */
 	const rec_t*	rec,		/* in: record to push; must
 					be protected by a page latch */
-	const ulint*	offsets)	/* in: rec_get_offsets() */
+	const ulint*	offsets,	/* in: rec_get_offsets() */
+	ulint		start_field_no,	/* in: start from this field */
+	byte*		remainder_buf)	/* in: if start_field_no !=0,
+					where to take prev fields */
 {
 	byte*	buf;
 	ulint	i;
@@ -3174,12 +3179,43 @@ row_sel_push_cache_row_for_mysql(
 	ut_ad(prebuilt->fetch_cache_first == 0);
 
 	if (UNIV_UNLIKELY(!row_sel_store_mysql_rec(
-				  prebuilt->fetch_cache[
+				prebuilt->fetch_cache[
 					  prebuilt->n_fetch_cached],
-				  prebuilt, rec, offsets))) {
+				prebuilt,
+				rec,
+				offsets,
+				start_field_no,
+				prebuilt->n_template))) {
 		ut_error;
 	}
 
+	if (start_field_no) {
+
+		for (i=0; i < start_field_no; i++) {
+			register		ulint offs;
+			mysql_row_templ_t*	templ;
+                        register byte *         null_byte;
+
+			templ = prebuilt->mysql_template + i;
+
+			if (templ->mysql_null_bit_mask) {
+				offs = templ->mysql_null_byte_offset;
+
+                                null_byte= prebuilt->fetch_cache[
+                                             prebuilt->n_fetch_cached]+offs;
+                                (*null_byte)&= ~templ->mysql_null_bit_mask;
+                                (*null_byte)|= (*(remainder_buf + offs) & 
+                                              templ->mysql_null_bit_mask);  
+			}
+
+			offs = templ->mysql_col_offset;
+			memcpy(prebuilt->fetch_cache[prebuilt->n_fetch_cached]
+			       + offs,
+			       remainder_buf + offs,
+			       templ->mysql_col_len);
+		}
+	}
+
 	prebuilt->n_fetch_cached++;
 }
 
@@ -3318,6 +3354,10 @@ row_search_for_mysql(
 	mem_heap_t*	heap				= NULL;
 	ulint		offsets_[REC_OFFS_NORMAL_SIZE];
 	ulint*		offsets				= offsets_;
+        /*psergey-mrr:*/
+	ibool		some_fields_in_buffer;
+	ibool		get_clust_rec			= 0;
+        /*:psergey-mrr*/
 
 	rec_offs_init(offsets_);
 
@@ -3570,7 +3610,8 @@ row_search_for_mysql(
 				mtr_commit(&mtr). */
 
 				if (!row_sel_store_mysql_rec(buf, prebuilt,
-							     rec, offsets)) {
+						rec, offsets, 0,
+						prebuilt->n_template)) {
 					err = DB_TOO_BIG_RECORD;
 
 					/* We let the main loop to do the
@@ -4169,8 +4210,11 @@ no_gap_lock:
 			information via the clustered index record. */
 
 			ut_ad(index != clust_index);
-
-			goto requires_clust_rec;
+			/*psergey-mrr:*/
+                        get_clust_rec = TRUE;
+			goto idx_cond_check;
+			/**goto requires_clust_rec;**/
+			/*:psergey-mrr*/
 		}
 	}
 
@@ -4214,12 +4258,30 @@ no_gap_lock:
 		goto next_rec;
 	}
 
+
+idx_cond_check:
+        if (prebuilt->idx_cond_func)
+        {
+          int res;
+          ut_ad(prebuilt->template_type != ROW_MYSQL_DUMMY_TEMPLATE);
+          offsets = rec_get_offsets(rec, index, offsets, ULINT_UNDEFINED, &heap);
+          row_sel_store_mysql_rec(buf, prebuilt, rec,
+                                  offsets, 0, prebuilt->n_index_fields);
+          res= prebuilt->idx_cond_func(prebuilt->idx_cond_func_arg);
+          if (res == 0)
+            goto next_rec;
+          if (res == 2)
+          {
+            err = DB_RECORD_NOT_FOUND;
+            goto idx_cond_failed;
+          }
+        }
+
 	/* Get the clustered index record if needed, if we did not do the
 	search using the clustered index. */
+	if (get_clust_rec || (index != clust_index
+			      && prebuilt->need_to_access_clustered)) {
 
-	if (index != clust_index && prebuilt->need_to_access_clustered) {
-
-requires_clust_rec:
 		/* We use a 'goto' to the preceding label if a consistent
 		read of a secondary index record requires us to look up old
 		versions of the associated clustered index record. */
@@ -4322,9 +4384,15 @@ requires_clust_rec:
 		are BLOBs in the fields to be fetched. In HANDLER we do
 		not cache rows because there the cursor is a scrollable
 		cursor. */
-
-		row_sel_push_cache_row_for_mysql(prebuilt, result_rec,
-						 offsets);
+		some_fields_in_buffer = (index != clust_index
+					 && prebuilt->idx_cond_func);
+
+		row_sel_push_cache_row_for_mysql(prebuilt,
+						 result_rec,
+						 offsets,
+						 some_fields_in_buffer?
+						 prebuilt->n_index_fields : 0,
+						 buf);
 		if (prebuilt->n_fetch_cached == MYSQL_FETCH_CACHE_SIZE) {
 
 			goto got_row;
@@ -4340,7 +4408,10 @@ requires_clust_rec:
 					rec_offs_extra_size(offsets) + 4);
 		} else {
 			if (!row_sel_store_mysql_rec(buf, prebuilt,
-						     result_rec, offsets)) {
+						   result_rec, offsets,
+						   prebuilt->idx_cond_func?
+						   prebuilt->n_index_fields: 0,
+						   prebuilt->n_template)) {
 				err = DB_TOO_BIG_RECORD;
 
 				goto lock_wait_or_error;
@@ -4368,6 +4439,9 @@ got_row:
 	HANDLER command where the user can move the cursor with PREV or NEXT
 	even after a unique search. */
 
+	err = DB_SUCCESS;
+
+idx_cond_failed:
 	if (!unique_search_from_clust_index
 	    || prebuilt->select_lock_type != LOCK_NONE
 	    || prebuilt->used_in_HANDLER) {
@@ -4377,12 +4451,11 @@ got_row:
 		btr_pcur_store_position(pcur, &mtr);
 	}
 
-	err = DB_SUCCESS;
-
 	goto normal_return;
 
 next_rec:
 	/* Reset the old and new "did semi-consistent read" flags. */
+	get_clust_rec = FALSE;
 	if (UNIV_UNLIKELY(prebuilt->row_read_type
 			  == ROW_READ_DID_SEMI_CONSISTENT)) {
 		prebuilt->row_read_type = ROW_READ_TRY_SEMI_CONSISTENT;