Added page_range to records_in_range() to improve range statistics

Prototype change:
-  virtual ha_rows records_in_range(uint inx, key_range *min_key,
-                                   key_range *max_key)
+  virtual ha_rows records_in_range(uint inx, const key_range *min_key,
+                                   const key_range *max_key,
+                                   page_range *res)

The handler can ignore the page_range parameter. In the case the handler
updates the parameter, the optimizer can deduce the following:
- If previous range's last key is on the same block as next range's first
  key
- If the current key range is in one block
- We can also assume that the first and last block read are cached!
  This can be used for a better calculation of IO seeks when we
  estimate the cost of a range index scan.

The parameter is fully implemented for MyISAM, Aria and InnoDB.
A separate patch will update handler::multi_range_read_info_const() to
take the benefits of this change and also remove the double
records_in_range() calls that are not anymore needed.

53 files changed, 290 insertions, 174 deletions

diff --git a/storage/connect/ha_connect.cc b/storage/connect/ha_connect.cc
index 489155a39a2..8fa39cc2651 100644
--- a/storage/connect/ha_connect.cc
+++ b/storage/connect/ha_connect.cc
@@ -5232,8 +5232,11 @@ int ha_connect::rename_table(const char *from, const char *to)
   @see
   check_quick_keys() in opt_range.cc
 */
-ha_rows ha_connect::records_in_range(uint inx, key_range *min_key,
-                                               key_range *max_key)
+ha_rows ha_connect::records_in_range(uint inx,
+                                     const key_range *min_key,
+                                     const key_range *max_key,
+                                     page_range *pages)
+
 {
   ha_rows rows;
   DBUG_ENTER("ha_connect::records_in_range");
diff --git a/storage/connect/ha_connect.h b/storage/connect/ha_connect.h
index 06de375ef58..09e8c42b443 100644
--- a/storage/connect/ha_connect.h
+++ b/storage/connect/ha_connect.h
@@ -461,8 +461,8 @@ int index_prev(uchar *buf);
   int start_stmt(THD *thd, thr_lock_type lock_type);
   int external_lock(THD *thd, int lock_type);                   ///< required
   int delete_all_rows(void);
-  ha_rows records_in_range(uint inx, key_range *min_key,
-                           key_range *max_key);
+  ha_rows records_in_range(uint inx, const key_range *start_key,
+                           const key_range *end_key, page_range *pages);
   /**
     These methods can be overridden, but their default implementation
     provide useful functionality.
diff --git a/storage/example/ha_example.cc b/storage/example/ha_example.cc
index 65b567aaac0..3bfe1e5e27b 100644
--- a/storage/example/ha_example.cc
+++ b/storage/example/ha_example.cc
@@ -839,6 +839,10 @@ int ha_example::delete_table(const char *name)
   @brief
   Given a starting key and an ending key, estimate the number of rows that
   will exist between the two keys.
+  The handler can also optionally update the 'pages' parameter with the page
+  number that contains the min and max keys. This will help the optimizer
+  to know if two ranges are partly on the same pages and if the min and
+  max key are on the same page.
 
   @details
   end_key may be empty, in which case determine if start_key matches any rows.
@@ -848,8 +852,10 @@ int ha_example::delete_table(const char *name)
   @see
   check_quick_keys() in opt_range.cc
 */
-ha_rows ha_example::records_in_range(uint inx, key_range *min_key,
-                                     key_range *max_key)
+ha_rows ha_example::records_in_range(uint inx,
+                                     const key_range *min_key,
+                                     const key_range *max_key,
+                                     page_range *pages)
 {
   DBUG_ENTER("ha_example::records_in_range");
   DBUG_RETURN(10);                         // low number to force index usage
diff --git a/storage/example/ha_example.h b/storage/example/ha_example.h
index 0a08e871461..2d3fa6d4216 100644
--- a/storage/example/ha_example.h
+++ b/storage/example/ha_example.h
@@ -242,8 +242,8 @@ public:
   int extra(enum ha_extra_function operation);
   int external_lock(THD *thd, int lock_type);                   ///< required
   int delete_all_rows(void);
-  ha_rows records_in_range(uint inx, key_range *min_key,
-                           key_range *max_key);
+  ha_rows records_in_range(uint inx, const key_range *min_key,
+                           const key_range *max_key, page_range *pages);
   int delete_table(const char *from);
   int create(const char *name, TABLE *form,
              HA_CREATE_INFO *create_info);                      ///< required
diff --git a/storage/federated/ha_federated.cc b/storage/federated/ha_federated.cc
index b87c5c39c37..00168791d35 100644
--- a/storage/federated/ha_federated.cc
+++ b/storage/federated/ha_federated.cc
@@ -1603,8 +1603,10 @@ static int free_share(FEDERATED_SHARE *share)
 }
 
 
-ha_rows ha_federated::records_in_range(uint inx, key_range *start_key,
-                                       key_range *end_key)
+ha_rows ha_federated::records_in_range(uint inx,
+                                       const key_range *start_key,
+                                       const key_range *end_key,
+                                       page_range *pages)
 {
   /*
 
diff --git a/storage/federated/ha_federated.h b/storage/federated/ha_federated.h
index 3beb2ca9570..080d0ebd5f0 100644
--- a/storage/federated/ha_federated.h
+++ b/storage/federated/ha_federated.h
@@ -249,8 +249,10 @@ public:
   int truncate();
   int create(const char *name, TABLE *form,
              HA_CREATE_INFO *create_info);                      //required
-  ha_rows records_in_range(uint inx, key_range *start_key,
-                                   key_range *end_key);
+  ha_rows records_in_range(uint inx,
+                           const key_range *start_key,
+                           const key_range *end_key,
+                           page_range *pages);
   uint8 table_cache_type() { return HA_CACHE_TBL_NOCACHE; }
 
   THR_LOCK_DATA **store_lock(THD *thd, THR_LOCK_DATA **to,
diff --git a/storage/federatedx/ha_federatedx.cc b/storage/federatedx/ha_federatedx.cc
index b76391dae4e..fef349bbaaf 100644
--- a/storage/federatedx/ha_federatedx.cc
+++ b/storage/federatedx/ha_federatedx.cc
@@ -1731,8 +1731,10 @@ static void free_share(federatedx_txn *txn, FEDERATEDX_SHARE *share)
 }
 
 
-ha_rows ha_federatedx::records_in_range(uint inx, key_range *start_key,
-                                       key_range *end_key)
+ha_rows ha_federatedx::records_in_range(uint inx,
+                                        const key_range *start_key,
+                                        const key_range *end_key,
+                                        page_range *pages)
 {
   /*
 
diff --git a/storage/federatedx/ha_federatedx.h b/storage/federatedx/ha_federatedx.h
index 1870a83d13d..a62456e1c33 100644
--- a/storage/federatedx/ha_federatedx.h
+++ b/storage/federatedx/ha_federatedx.h
@@ -435,8 +435,8 @@ public:
   int delete_all_rows(void);
   int create(const char *name, TABLE *form,
              HA_CREATE_INFO *create_info);                      //required
-  ha_rows records_in_range(uint inx, key_range *start_key,
-                                   key_range *end_key);
+  ha_rows records_in_range(uint inx, const key_range *start_key,
+                           const key_range *end_key, page_range *pages);
   uint8 table_cache_type() { return HA_CACHE_TBL_NOCACHE; }
 
   THR_LOCK_DATA **store_lock(THD *thd, THR_LOCK_DATA **to,
diff --git a/storage/heap/ha_heap.cc b/storage/heap/ha_heap.cc
index ccfd6c69956..34e2799f71c 100644
--- a/storage/heap/ha_heap.cc
+++ b/storage/heap/ha_heap.cc
@@ -579,8 +579,8 @@ int ha_heap::rename_table(const char * from, const char * to)
 }
 
 
-ha_rows ha_heap::records_in_range(uint inx, key_range *min_key,
-                                  key_range *max_key)
+ha_rows ha_heap::records_in_range(uint inx, const key_range *min_key,
+                                  const key_range *max_key, page_range *pages)
 {
   KEY *key=table->key_info+inx;
   if (key->algorithm == HA_KEY_ALG_BTREE)
diff --git a/storage/heap/ha_heap.h b/storage/heap/ha_heap.h
index 370906bd1f6..67b9ac9524a 100644
--- a/storage/heap/ha_heap.h
+++ b/storage/heap/ha_heap.h
@@ -102,7 +102,8 @@ public:
   int disable_indexes(uint mode);
   int enable_indexes(uint mode);
   int indexes_are_disabled(void);
-  ha_rows records_in_range(uint inx, key_range *min_key, key_range *max_key);
+  ha_rows records_in_range(uint inx, const key_range *start_key,
+                           const key_range *end_key, page_range *pages);
   int delete_table(const char *from);
   void drop_table(const char *name);
   int rename_table(const char * from, const char * to);
diff --git a/storage/heap/hp_hash.c b/storage/heap/hp_hash.c
index 00967dca52f..a0139151730 100644
--- a/storage/heap/hp_hash.c
+++ b/storage/heap/hp_hash.c
@@ -55,8 +55,9 @@ static ulong hp_hashnr(HP_KEYDEF *keydef, const uchar *key);
 			the range.
 */
 
-ha_rows hp_rb_records_in_range(HP_INFO *info, int inx,  key_range *min_key,
-                               key_range *max_key)
+ha_rows hp_rb_records_in_range(HP_INFO *info, int inx,
+                               const key_range *min_key,
+                               const key_range *max_key)
 {
   ha_rows start_pos, end_pos;
   HP_KEYDEF *keyinfo= info->s->keydef + inx;
diff --git a/storage/innobase/btr/btr0cur.cc b/storage/innobase/btr/btr0cur.cc
index ba5979ff1c6..8a547669894 100644
--- a/storage/innobase/btr/btr0cur.cc
+++ b/storage/innobase/btr/btr0cur.cc
@@ -6135,10 +6135,8 @@ static const ha_rows	rows_in_range_arbitrary_ret_val = 10;
 
 /** Estimates the number of rows in a given index range.
 @param[in]	index		index
-@param[in]	tuple1		range start, may also be empty tuple
-@param[in]	mode1		search mode for range start
-@param[in]	tuple2		range end, may also be empty tuple
-@param[in]	mode2		search mode for range end
+@param[in]	tuple1		range start
+@param[in]	tuple2		range end
 @param[in]	nth_attempt	if the tree gets modified too much while
 we are trying to analyze it, then we will retry (this function will call
 itself, incrementing this parameter)
@@ -6151,10 +6149,8 @@ static
 ha_rows
 btr_estimate_n_rows_in_range_low(
 	dict_index_t*	index,
-	const dtuple_t*	tuple1,
-	page_cur_mode_t	mode1,
-	const dtuple_t*	tuple2,
-	page_cur_mode_t	mode2,
+	btr_pos_t*	tuple1,
+	btr_pos_t*	tuple2,
 	unsigned	nth_attempt)
 {
 	btr_path_t	path1[BTR_PATH_ARRAY_N_SLOTS];
@@ -6170,6 +6166,7 @@ btr_estimate_n_rows_in_range_low(
 	ulint		i;
 	mtr_t		mtr;
 	ha_rows		table_n_rows;
+        page_cur_mode_t mode2= tuple2->mode;
 
 	table_n_rows = dict_table_get_n_rows(index->table);
 
@@ -6188,9 +6185,10 @@ btr_estimate_n_rows_in_range_low(
 
 	bool	should_count_the_left_border;
 
-	if (dtuple_get_n_fields(tuple1) > 0) {
+	if (dtuple_get_n_fields(tuple1->tuple) > 0) {
 
-		btr_cur_search_to_nth_level(index, 0, tuple1, mode1,
+              btr_cur_search_to_nth_level(index, 0, tuple1->tuple,
+                                            tuple1->mode,
 					    BTR_SEARCH_LEAF | BTR_ESTIMATE,
 					    &cursor, 0,
 					    __FILE__, __LINE__, &mtr);
@@ -6230,6 +6228,8 @@ btr_estimate_n_rows_in_range_low(
 		should_count_the_left_border = false;
 	}
 
+        tuple1->page_id= cursor.page_cur.block->page.id;
+
 	mtr_commit(&mtr);
 
 	if (!index->is_readable()) {
@@ -6242,9 +6242,10 @@ btr_estimate_n_rows_in_range_low(
 
 	bool	should_count_the_right_border;
 
-	if (dtuple_get_n_fields(tuple2) > 0) {
+	if (dtuple_get_n_fields(tuple2->tuple) > 0) {
 
-		btr_cur_search_to_nth_level(index, 0, tuple2, mode2,
+		btr_cur_search_to_nth_level(index, 0, tuple2->tuple,
+                                            mode2,
 					    BTR_SEARCH_LEAF | BTR_ESTIMATE,
 					    &cursor, 0,
 					    __FILE__, __LINE__, &mtr);
@@ -6256,7 +6257,7 @@ btr_estimate_n_rows_in_range_low(
 		should_count_the_right_border
 			= (mode2 == PAGE_CUR_LE /* if the range is '<=' */
 			   /* and the record was found */
-			   && cursor.low_match >= dtuple_get_n_fields(tuple2))
+			   && cursor.low_match >= dtuple_get_n_fields(tuple2->tuple))
 			|| (mode2 == PAGE_CUR_L /* or if the range is '<' */
 			    /* and there are any records to match the criteria,
 			    i.e. if the minimum record on the tree is 5 and
@@ -6300,6 +6301,8 @@ btr_estimate_n_rows_in_range_low(
 		should_count_the_right_border = false;
 	}
 
+        tuple2->page_id= cursor.page_cur.block->page.id;
+
 	mtr_commit(&mtr);
 
 	/* We have the path information for the range in path1 and path2 */
@@ -6438,8 +6441,8 @@ btr_estimate_n_rows_in_range_low(
 				}
 
 				return btr_estimate_n_rows_in_range_low(
-					index, tuple1, mode1,
-					tuple2, mode2, nth_attempt + 1);
+                                       index, tuple1, tuple2,
+                                       nth_attempt + 1);
 			}
 
 			diverged = true;
@@ -6510,13 +6513,11 @@ btr_estimate_n_rows_in_range_low(
 ha_rows
 btr_estimate_n_rows_in_range(
 	dict_index_t*	index,
-	const dtuple_t*	tuple1,
-	page_cur_mode_t	mode1,
-	const dtuple_t*	tuple2,
-	page_cur_mode_t	mode2)
+        btr_pos_t       *tuple1,
+        btr_pos_t       *tuple2)
 {
 	return btr_estimate_n_rows_in_range_low(
-		index, tuple1, mode1, tuple2, mode2, 1);
+		index, tuple1, tuple2, 1);
 }
 
 /*******************************************************************//**
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index 2d3471fd65c..9c5db57c661 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -13936,10 +13936,11 @@ ha_rows
 ha_innobase::records_in_range(
 /*==========================*/
 	uint			keynr,		/*!< in: index number */
-	key_range		*min_key,	/*!< in: start key value of the
+	const key_range		*min_key,	/*!< in: start key value of the
 						range, may also be 0 */
-	key_range		*max_key)	/*!< in: range end key val, may
+	const key_range		*max_key,	/*!< in: range end key val, may
 						also be 0 */
+        page_range              *pages)
 {
 	KEY*		key;
 	dict_index_t*	index;
@@ -14028,8 +14029,12 @@ ha_innobase::records_in_range(
 			n_rows = rtr_estimate_n_rows_in_range(
 				index, range_start, mode1);
 		} else {
+                        btr_pos_t tuple1(range_start, mode1, pages->first_page);
+                        btr_pos_t tuple2(range_end,   mode2, pages->last_page);
 			n_rows = btr_estimate_n_rows_in_range(
-				index, range_start, mode1, range_end, mode2);
+                                 index, &tuple1, &tuple2);
+                        pages->first_page= tuple1.page_id.raw();
+                        pages->last_page=  tuple2.page_id.raw();
 		}
 	} else {
 
diff --git a/storage/innobase/handler/ha_innodb.h b/storage/innobase/handler/ha_innodb.h
index 980c5b06ea5..d48c14da970 100644
--- a/storage/innobase/handler/ha_innodb.h
+++ b/storage/innobase/handler/ha_innodb.h
@@ -184,9 +184,10 @@ public:
 	int start_stmt(THD *thd, thr_lock_type lock_type) override;
 
 	ha_rows records_in_range(
-		uint			inx,
-		key_range*		min_key,
-		key_range*		max_key) override;
+                uint                    inx,
+                const key_range*        min_key,
+                const key_range*        max_key,
+                page_range*             pages) override;
 
 	ha_rows estimate_rows_upper_bound() override;
 
diff --git a/storage/innobase/include/btr0cur.h b/storage/innobase/include/btr0cur.h
index 0a45c3bb45f..9d5c9283e3b 100644
--- a/storage/innobase/include/btr0cur.h
+++ b/storage/innobase/include/btr0cur.h
@@ -555,20 +555,41 @@ btr_cur_pessimistic_delete(
 @param[in,out]	mtr	mini-transaction */
 void btr_cur_node_ptr_delete(btr_cur_t* parent, mtr_t* mtr)
 	MY_ATTRIBUTE((nonnull));
+/***********************************************************//**
+Parses a redo log record of updating a record in-place.
+@return end of log record or NULL */
+byte*
+btr_cur_parse_update_in_place(
+/*==========================*/
+	byte*		ptr,	/*!< in: buffer */
+	byte*		end_ptr,/*!< in: buffer end */
+	page_t*		page,	/*!< in/out: page or NULL */
+	page_zip_des_t*	page_zip,/*!< in/out: compressed page, or NULL */
+	dict_index_t*	index);	/*!< in: index corresponding to page */
+/** Arguments to btr_estimate_n_rows_in_range */
+struct btr_pos_t
+{
+  btr_pos_t(dtuple_t *arg_tuple,
+            page_cur_mode_t arg_mode,
+            page_id_t arg_page_id)
+  :tuple(arg_tuple), mode(arg_mode), page_id(arg_page_id)
+  {}
+
+  dtuple_t*       tuple;       /* Range start or end. May be NULL */
+  page_cur_mode_t mode;        /* search mode for range */
+  page_id_t       page_id;     /* Out: Page where we found the tuple */
+};
+
 /** Estimates the number of rows in a given index range.
 @param[in]	index	index
-@param[in]	tuple1	range start, may also be empty tuple
-@param[in]	mode1	search mode for range start
-@param[in]	tuple2	range end, may also be empty tuple
-@param[in]	mode2	search mode for range end
+@param[in/out]	range_start
+@param[in/out]	range_ end
 @return estimated number of rows */
 ha_rows
 btr_estimate_n_rows_in_range(
 	dict_index_t*	index,
-	const dtuple_t*	tuple1,
-	page_cur_mode_t	mode1,
-	const dtuple_t*	tuple2,
-	page_cur_mode_t	mode2);
+        btr_pos_t*      range_start,
+        btr_pos_t*      range_end);
 
 /*******************************************************************//**
 Estimates the number of different key values in a given index, for
diff --git a/storage/innobase/include/buf0types.h b/storage/innobase/include/buf0types.h
index 124a7f1854f..34146ccdfa5 100644
--- a/storage/innobase/include/buf0types.h
+++ b/storage/innobase/include/buf0types.h
@@ -141,6 +141,7 @@ public:
     ut_ad(page_no <= 0xFFFFFFFFU);
   }
 
+  page_id_t(ulonglong id) : m_id(id) {}
   bool operator==(const page_id_t& rhs) const { return m_id == rhs.m_id; }
   bool operator!=(const page_id_t& rhs) const { return m_id != rhs.m_id; }
 
@@ -169,6 +170,7 @@ public:
   /** Set the FIL_NULL for the space and page_no */
   void set_corrupt_id() { m_id= ~uint64_t{0}; }
 
+  ulonglong raw() { return m_id; }
 private:
   /** The page identifier */
   uint64_t m_id;
diff --git a/storage/maria/ha_maria.cc b/storage/maria/ha_maria.cc
index 737a737f791..eb197f59925 100644
--- a/storage/maria/ha_maria.cc
+++ b/storage/maria/ha_maria.cc
@@ -3296,6 +3296,8 @@ void ha_maria::get_auto_increment(ulonglong offset, ulonglong increment,
     inx                 Index to use
     min_key             Start of range.  Null pointer if from first key
     max_key             End of range. Null pointer if to last key
+    pages               Store first and last page for the range in case of
+                        b-trees. In other cases it's not touched.
 
   NOTES
     min_key.flag can have one of the following values:
@@ -3313,11 +3315,12 @@ void ha_maria::get_auto_increment(ulonglong offset, ulonglong increment,
                         the range.
 */
 
-ha_rows ha_maria::records_in_range(uint inx, key_range *min_key,
-                                   key_range *max_key)
+ha_rows ha_maria::records_in_range(uint inx, const key_range *min_key,
+                                   const key_range *max_key, page_range *pages)
 {
   register_handler(file);
-  return (ha_rows) maria_records_in_range(file, (int) inx, min_key, max_key);
+  return (ha_rows) maria_records_in_range(file, (int) inx, min_key, max_key,
+                                          pages);
 }
 
 
diff --git a/storage/maria/ha_maria.h b/storage/maria/ha_maria.h
index 691d1c9747b..d24dee38135 100644
--- a/storage/maria/ha_maria.h
+++ b/storage/maria/ha_maria.h
@@ -124,7 +124,9 @@ public:
   int indexes_are_disabled(void);
   void start_bulk_insert(ha_rows rows, uint flags);
   int end_bulk_insert();
-  ha_rows records_in_range(uint inx, key_range * min_key, key_range * max_key);
+  ha_rows records_in_range(uint inx, const key_range *min_key,
+                           const key_range *max_key,
+                           page_range *pages);
   void update_create_info(HA_CREATE_INFO * create_info);
   int create(const char *name, TABLE * form, HA_CREATE_INFO * create_info);
   THR_LOCK_DATA **store_lock(THD * thd, THR_LOCK_DATA ** to,
diff --git a/storage/maria/ma_range.c b/storage/maria/ma_range.c
index d4e449a748e..442adc35858 100644
--- a/storage/maria/ma_range.c
+++ b/storage/maria/ma_range.c
@@ -22,8 +22,9 @@
 #include "ma_rt_index.h"
 
 static ha_rows _ma_record_pos(MARIA_HA *,const uchar *, key_part_map,
-			      enum ha_rkey_function);
-static double _ma_search_pos(MARIA_HA *, MARIA_KEY *, uint32, my_off_t);
+			      enum ha_rkey_function, ulonglong *);
+static double _ma_search_pos(MARIA_HA *, MARIA_KEY *, uint32, my_off_t,
+                             ulonglong *page);
 static uint _ma_keynr(MARIA_PAGE *page, uchar *keypos, uint *ret_max_key);
 
 
@@ -43,8 +44,9 @@ static uint _ma_keynr(MARIA_PAGE *page, uchar *keypos, uint *ret_max_key);
      @retval number        Estimated number of rows
 */
 
-ha_rows maria_records_in_range(MARIA_HA *info, int inx, key_range *min_key,
-                            key_range *max_key)
+ha_rows maria_records_in_range(MARIA_HA *info, int inx,
+                               const key_range *min_key,
+                               const key_range *max_key, page_range *pages)
 {
   ha_rows start_pos,end_pos,res;
   MARIA_SHARE *share= info->s;
@@ -96,11 +98,11 @@ ha_rows maria_records_in_range(MARIA_HA *info, int inx, key_range *min_key,
   default:
     start_pos= (min_key ?
                 _ma_record_pos(info, min_key->key, min_key->keypart_map,
-                               min_key->flag) :
+                               min_key->flag, &pages->first_page) :
                 (ha_rows) 0);
     end_pos=   (max_key ?
                 _ma_record_pos(info, max_key->key, max_key->keypart_map,
-                               max_key->flag) :
+                               max_key->flag, &pages->last_page) :
                 info->state->records + (ha_rows) 1);
     res= (end_pos < start_pos ? (ha_rows) 0 :
           (end_pos == start_pos ? (ha_rows) 1 : end_pos-start_pos));
@@ -128,7 +130,8 @@ ha_rows maria_records_in_range(MARIA_HA *info, int inx, key_range *min_key,
 
 static ha_rows _ma_record_pos(MARIA_HA *info, const uchar *key_data,
                               key_part_map keypart_map,
-			      enum ha_rkey_function search_flag)
+                              enum ha_rkey_function search_flag,
+                              ulonglong *final_page)
 {
   uint inx= (uint) info->lastinx;
   uint32 nextflag;
@@ -185,7 +188,7 @@ static ha_rows _ma_record_pos(MARIA_HA *info, const uchar *key_data,
   */
   pos= _ma_search_pos(info, &key,
                       nextflag | SEARCH_SAVE_BUFF | SEARCH_UPDATE,
-                      info->s->state.key_root[inx]);
+                      info->s->state.key_root[inx], final_page);
   if (pos >= 0.0)
   {
     DBUG_PRINT("exit",("pos: %ld",(ulong) (pos*info->state->records)));
@@ -206,7 +209,8 @@ static ha_rows _ma_record_pos(MARIA_HA *info, const uchar *key_data,
 */
 
 static double _ma_search_pos(MARIA_HA *info, MARIA_KEY *key,
-			     uint32 nextflag, my_off_t pos)
+                             uint32 nextflag, my_off_t pos,
+                             ulonglong *final_page)
 {
   int flag;
   uint keynr, UNINIT_VAR(max_keynr);
@@ -224,6 +228,7 @@ static double _ma_search_pos(MARIA_HA *info, MARIA_KEY *key,
                         PAGECACHE_LOCK_LEFT_UNLOCKED, DFLT_INIT_HITS,
                         info->buff, 1))
     goto err;
+  *final_page= pos;
   flag= (*keyinfo->bin_search)(key, &page, nextflag, &keypos,
                                info->lastkey_buff, &after_key);
   keynr= _ma_keynr(&page, keypos, &max_keynr);
@@ -240,7 +245,8 @@ static double _ma_search_pos(MARIA_HA *info, MARIA_KEY *key,
     if (! page.node)
       offset= 0.0;
     else if ((offset= _ma_search_pos(info, key, nextflag,
-                                     _ma_kpos(page.node,keypos))) < 0)
+                                     _ma_kpos(page.node,keypos),
+                                     final_page)) < 0)
       DBUG_RETURN(offset);
   }
   else
@@ -269,7 +275,8 @@ static double _ma_search_pos(MARIA_HA *info, MARIA_KEY *key,
         Matches keynr + [0-1]
       */
       if ((offset= _ma_search_pos(info, key, SEARCH_FIND,
-                                  _ma_kpos(page.node,keypos))) < 0)
+                                  _ma_kpos(page.node,keypos),
+                                  final_page)) < 0)
 	DBUG_RETURN(offset);			/* Read error */
     }
   }
diff --git a/storage/maria/ma_rt_test.c b/storage/maria/ma_rt_test.c
index 57c4eea4d9c..12f35af7cc1 100644
--- a/storage/maria/ma_rt_test.c
+++ b/storage/maria/ma_rt_test.c
@@ -142,6 +142,7 @@ static int run_test(const char *filename)
   uchar read_record[MAX_REC_LENGTH];
   int upd= 10;
   ha_rows hrows;
+  page_range pages;
 
   bzero(&uniquedef, sizeof(uniquedef));
   bzero(&create_info, sizeof(create_info));
@@ -427,7 +428,7 @@ static int run_test(const char *filename)
   range.key= record+1;
   range.length= 1000;                           /* Big enough */
   range.flag= HA_READ_MBR_INTERSECT;
-  hrows= maria_records_in_range(file,0, &range, (key_range*) 0);
+  hrows= maria_records_in_range(file,0, &range, (key_range*) 0, &pages);
   if (!silent)
     printf("     %ld rows\n", (long) hrows);
 
diff --git a/storage/maria/ma_sp_test.c b/storage/maria/ma_sp_test.c
index 21adb33968e..290a94872b5 100644
--- a/storage/maria/ma_sp_test.c
+++ b/storage/maria/ma_sp_test.c
@@ -71,6 +71,7 @@ int run_test(const char *filename)
   uchar read_record[MAX_REC_LENGTH];
   int upd=10;
   ha_rows hrows;
+  page_range pages;
 
   /* Define a column for NULLs and DEL markers*/
 
@@ -258,7 +259,7 @@ int run_test(const char *filename)
   max_range.key= record+1;
   max_range.length= 1000;                       /* Big enough */
   max_range.flag= HA_READ_KEY_EXACT;
-  hrows= maria_records_in_range(file,0, &min_range, &max_range);
+  hrows= maria_records_in_range(file,0, &min_range, &max_range, &pages);
   printf("     %ld rows\n", (long) hrows);
 
   if (maria_close(file)) goto err;
diff --git a/storage/maria/ma_test2.c b/storage/maria/ma_test2.c
index 956ac2d2f94..18ab2543bb0 100644
--- a/storage/maria/ma_test2.c
+++ b/storage/maria/ma_test2.c
@@ -69,6 +69,7 @@ int main(int argc, char *argv[])
   char *blob_buffer;
   MARIA_CREATE_INFO create_info;
   char filename[FN_REFLEN];
+  page_range pages;
 
 #ifdef SAFE_MUTEX
   safe_mutex_deadlock_detector= 1;
@@ -722,7 +723,8 @@ int main(int argc, char *argv[])
     max_key.keypart_map= HA_WHOLE_KEY;
     max_key.flag= HA_READ_AFTER_KEY;
 
-    range_records= maria_records_in_range(file,(int) i, &min_key, &max_key);
+    range_records= maria_records_in_range(file,(int) i, &min_key, &max_key,
+                                          &pages);
     if (range_records < info.records*8/10 ||
 	range_records > info.records*12/10)
     {
@@ -756,7 +758,8 @@ int main(int argc, char *argv[])
       max_key.key= key2;
       max_key.keypart_map= HA_WHOLE_KEY;
       max_key.flag= HA_READ_BEFORE_KEY;
-      range_records= maria_records_in_range(file, 0, &min_key, &max_key);
+      range_records= maria_records_in_range(file, 0, &min_key, &max_key,
+                                            &pages);
       records=0;
       for (j++ ; j < k ; j++)
 	records+=key1[j];
diff --git a/storage/mroonga/ha_mroonga.cpp b/storage/mroonga/ha_mroonga.cpp
index 49208fc51bb..728fd8acd7c 100644
--- a/storage/mroonga/ha_mroonga.cpp
+++ b/storage/mroonga/ha_mroonga.cpp
@@ -7437,8 +7437,10 @@ uint ha_mroonga::max_supported_key_parts() const
   DBUG_RETURN(parts);
 }
 
-ha_rows ha_mroonga::wrapper_records_in_range(uint key_nr, key_range *range_min,
-                                             key_range *range_max)
+ha_rows ha_mroonga::wrapper_records_in_range(uint key_nr,
+                                             const key_range *range_min,
+                                             const key_range *range_max,
+                                             page_range *pages)
 {
   ha_rows row_count;
   MRN_DBUG_ENTER_METHOD();
@@ -7448,15 +7450,18 @@ ha_rows ha_mroonga::wrapper_records_in_range(uint key_nr, key_range *range_min,
   } else {
     MRN_SET_WRAP_SHARE_KEY(share, table->s);
     MRN_SET_WRAP_TABLE_KEY(this, table);
-    row_count = wrap_handler->records_in_range(key_nr, range_min, range_max);
+    row_count = wrap_handler->records_in_range(key_nr, range_min, range_max,
+                                               pages);
     MRN_SET_BASE_SHARE_KEY(share, table->s);
     MRN_SET_BASE_TABLE_KEY(this, table);
   }
   DBUG_RETURN(row_count);
 }
 
-ha_rows ha_mroonga::storage_records_in_range(uint key_nr, key_range *range_min,
-                                             key_range *range_max)
+ha_rows ha_mroonga::storage_records_in_range(uint key_nr,
+                                             const key_range *range_min,
+                                             const key_range *range_max,
+                                             page_range *pages)
 {
   MRN_DBUG_ENTER_METHOD();
   int flags = 0;
@@ -7569,8 +7574,8 @@ ha_rows ha_mroonga::storage_records_in_range(uint key_nr, key_range *range_min,
 }
 
 ha_rows ha_mroonga::generic_records_in_range_geo(uint key_nr,
-                                                 key_range *range_min,
-                                                 key_range *range_max)
+                                                 const key_range *range_min,
+                                                 const key_range *range_max)
 {
   MRN_DBUG_ENTER_METHOD();
   ha_rows row_count;
@@ -7604,15 +7609,17 @@ ha_rows ha_mroonga::generic_records_in_range_geo(uint key_nr,
   DBUG_RETURN(row_count);
 }
 
-ha_rows ha_mroonga::records_in_range(uint key_nr, key_range *range_min, key_range *range_max)
+ha_rows ha_mroonga::records_in_range(uint key_nr, const key_range *range_min,
+                                     const key_range *range_max,
+                                     page_range *pages)
 {
   MRN_DBUG_ENTER_METHOD();
   ha_rows row_count = 0;
   if (share->wrapper_mode)
   {
-    row_count = wrapper_records_in_range(key_nr, range_min, range_max);
+    row_count = wrapper_records_in_range(key_nr, range_min, range_max, pages);
   } else {
-    row_count = storage_records_in_range(key_nr, range_min, range_max);
+    row_count = storage_records_in_range(key_nr, range_min, range_max, pages);
   }
   DBUG_PRINT("info", ("mroonga: row_count=%" MRN_HA_ROWS_FORMAT, row_count));
   DBUG_RETURN(row_count);
diff --git a/storage/mroonga/ha_mroonga.hpp b/storage/mroonga/ha_mroonga.hpp
index fdcb1c4f316..1304cce4504 100644
--- a/storage/mroonga/ha_mroonga.hpp
+++ b/storage/mroonga/ha_mroonga.hpp
@@ -461,7 +461,8 @@ public:
   uint max_supported_key_length()      const mrn_override;
   uint max_supported_key_part_length() const mrn_override;
 
-  ha_rows records_in_range(uint inx, key_range *min_key, key_range *max_key) mrn_override;
+  ha_rows records_in_range(uint inx, const key_range *min_key,
+                           const key_range *max_key, page_range *pages) mrn_override;
   int index_init(uint idx, bool sorted) mrn_override;
   int index_end() mrn_override;
 #ifndef MRN_HANDLER_HAVE_HA_INDEX_READ_MAP
@@ -984,12 +985,14 @@ private:
   int storage_rnd_pos(uchar *buf, uchar *pos);
   void wrapper_position(const uchar *record);
   void storage_position(const uchar *record);
-  ha_rows wrapper_records_in_range(uint key_nr, key_range *range_min,
-                                   key_range *range_max);
-  ha_rows storage_records_in_range(uint key_nr, key_range *range_min,
-                                   key_range *range_max);
-  ha_rows generic_records_in_range_geo(uint key_nr, key_range *range_min,
-                                       key_range *range_max);
+  ha_rows wrapper_records_in_range(uint key_nr, const key_range *range_min,
+                                   const key_range *range_max,
+                                   page_range *pages);
+  ha_rows storage_records_in_range(uint key_nr, const key_range *range_min,
+                                   const key_range *range_max,
+                                   page_range *pages);
+  ha_rows generic_records_in_range_geo(uint key_nr, const key_range *range_min,
+                                       const key_range *range_max);
   int wrapper_index_init(uint idx, bool sorted);
   int storage_index_init(uint idx, bool sorted);
   int wrapper_index_end();
diff --git a/storage/myisam/ha_myisam.cc b/storage/myisam/ha_myisam.cc
index 69406177e1c..e6137fac482 100644
--- a/storage/myisam/ha_myisam.cc
+++ b/storage/myisam/ha_myisam.cc
@@ -2337,6 +2337,8 @@ void ha_myisam::get_auto_increment(ulonglong offset, ulonglong increment,
     inx			Index to use
     min_key		Start of range.  Null pointer if from first key
     max_key		End of range. Null pointer if to last key
+    pages               Store first and last page for the range in case of
+                        b-trees. In other cases it's not touched.
 
   NOTES
     min_key.flag can have one of the following values:
@@ -2354,10 +2356,12 @@ void ha_myisam::get_auto_increment(ulonglong offset, ulonglong increment,
 			the range.
 */
 
-ha_rows ha_myisam::records_in_range(uint inx, key_range *min_key,
-                                    key_range *max_key)
+ha_rows ha_myisam::records_in_range(uint inx, const key_range *min_key,
+                                    const key_range *max_key,
+                                    page_range *pages)
 {
-  return (ha_rows) mi_records_in_range(file, (int) inx, min_key, max_key);
+  return (ha_rows) mi_records_in_range(file, (int) inx, min_key, max_key,
+                                       pages);
 }
 
 
diff --git a/storage/myisam/ha_myisam.h b/storage/myisam/ha_myisam.h
index 0af4215e8f9..e36927a2455 100644
--- a/storage/myisam/ha_myisam.h
+++ b/storage/myisam/ha_myisam.h
@@ -114,7 +114,8 @@ class ha_myisam: public handler
   int indexes_are_disabled(void);
   void start_bulk_insert(ha_rows rows, uint flags);
   int end_bulk_insert();
-  ha_rows records_in_range(uint inx, key_range *min_key, key_range *max_key);
+  ha_rows records_in_range(uint inx, const key_range *min_key,
+                           const key_range *max_key, page_range *pages);
   void update_create_info(HA_CREATE_INFO *create_info);
   int create(const char *name, TABLE *form, HA_CREATE_INFO *create_info);
   THR_LOCK_DATA **store_lock(THD *thd, THR_LOCK_DATA **to,
diff --git a/storage/myisam/mi_range.c b/storage/myisam/mi_range.c
index 7292f3c83b8..54350f7a111 100644
--- a/storage/myisam/mi_range.c
+++ b/storage/myisam/mi_range.c
@@ -24,9 +24,9 @@
 #include "rt_index.h"
 
 static double _mi_record_pos(MI_INFO *, const uchar *, key_part_map,
-                             enum ha_rkey_function);
+                             enum ha_rkey_function, ulonglong *);
 static double _mi_search_pos(MI_INFO *,MI_KEYDEF *,uchar *, uint,uint,
-                             my_off_t,my_bool);
+                             my_off_t,my_bool, ulonglong *);
 static uint _mi_keynr(MI_INFO *info,MI_KEYDEF *,uchar *, uchar *,uint *);
 
 /*
@@ -48,7 +48,8 @@ static uint _mi_keynr(MI_INFO *info,MI_KEYDEF *,uchar *, uchar *,uint *);
 */
   
 ha_rows mi_records_in_range(MI_INFO *info, int inx,
-                            key_range *min_key, key_range *max_key)
+                            const key_range *min_key, const key_range *max_key,
+                            page_range *pages)
 {
   ha_rows res;
   double start_pos,end_pos,diff;
@@ -98,10 +99,12 @@ ha_rows mi_records_in_range(MI_INFO *info, int inx,
   case HA_KEY_ALG_BTREE:
   default:
     start_pos= (min_key ?_mi_record_pos(info, min_key->key,
-                                        min_key->keypart_map, min_key->flag)
+                                        min_key->keypart_map, min_key->flag,
+                                        &pages->first_page)
                         : (double) 0);
     end_pos=   (max_key ?  _mi_record_pos(info, max_key->key,
-                                          max_key->keypart_map, max_key->flag)
+                                          max_key->keypart_map, max_key->flag,
+                                          &pages->last_page)
 		        : (double) info->state->records);
     res= (end_pos < start_pos ? (ha_rows) 0 :
           (end_pos == start_pos ? (ha_rows) 1 : (ha_rows) (end_pos-start_pos)));
@@ -147,7 +150,8 @@ ha_rows mi_records_in_range(MI_INFO *info, int inx,
 
 static double _mi_record_pos(MI_INFO *info, const uchar *key,
                              key_part_map keypart_map,
-                             enum ha_rkey_function search_flag)
+                             enum ha_rkey_function search_flag,
+                             ulonglong *final_page)
 {
   uint inx=(uint) info->lastinx, nextflag, key_len;
   MI_KEYDEF *keyinfo=info->s->keyinfo+inx;
@@ -203,7 +207,8 @@ static double _mi_record_pos(MI_INFO *info, const uchar *key,
   */
   pos=_mi_search_pos(info,keyinfo,key_buff,key_len,
 		     nextflag | SEARCH_SAVE_BUFF | SEARCH_UPDATE,
-		     info->s->state.key_root[inx], TRUE);
+		     info->s->state.key_root[inx], TRUE,
+                     final_page);
   if (pos >= 0.0)
   {
     DBUG_PRINT("exit",("pos: %g",(pos*info->state->records)));
@@ -219,7 +224,8 @@ static double _mi_record_pos(MI_INFO *info, const uchar *key,
 static double _mi_search_pos(register MI_INFO *info,
 			     register MI_KEYDEF *keyinfo,
 			     uchar *key, uint key_len, uint nextflag,
-			     register my_off_t pos, my_bool last_in_level)
+			     register my_off_t pos, my_bool last_in_level,
+                             ulonglong *final_page)
 {
   int flag;
   uint nod_flag,keynr,UNINIT_VAR(max_keynr);
@@ -233,6 +239,7 @@ static double _mi_search_pos(register MI_INFO *info,
 
   if (!(buff=_mi_fetch_keypage(info,keyinfo,pos,DFLT_INIT_HITS,info->buff,1)))
     goto err;
+  *final_page= pos;
   flag=(*keyinfo->bin_search)(info,keyinfo,buff,key,key_len,nextflag,
 			      &keypos,info->lastkey, &after_key);
   nod_flag=mi_test_if_nod(buff);
@@ -251,7 +258,8 @@ static double _mi_search_pos(register MI_INFO *info,
       offset= 1.0;
     else if ((offset=_mi_search_pos(info,keyinfo,key,key_len,nextflag,
 				    _mi_kpos(nod_flag,keypos),
-                                    last_in_level && after_key)) < 0)
+                                    last_in_level && after_key,
+                                    final_page)) < 0)
       DBUG_RETURN(offset);
   }
   else
@@ -271,7 +279,8 @@ static double _mi_search_pos(register MI_INFO *info,
       */
       if ((offset=_mi_search_pos(info,keyinfo,key,key_len,SEARCH_FIND,
 				 _mi_kpos(nod_flag,keypos),
-                                 last_in_level && after_key)) < 0)
+                                 last_in_level && after_key,
+                                 final_page)) < 0)
 	DBUG_RETURN(offset);			/* Read error */
     }
   }
diff --git a/storage/myisam/mi_test2.c b/storage/myisam/mi_test2.c
index 5a66e958139..4b5039eea5b 100644
--- a/storage/myisam/mi_test2.c
+++ b/storage/myisam/mi_test2.c
@@ -67,6 +67,7 @@ int main(int argc, char *argv[])
   const char *filename;
   char *blob_buffer;
   MI_CREATE_INFO create_info;
+  page_range pages;
   MY_INIT(argv[0]);
 
   filename= "test2";
@@ -622,7 +623,8 @@ int main(int argc, char *argv[])
     max_key.keypart_map= HA_WHOLE_KEY;
     max_key.flag= HA_READ_AFTER_KEY;
 
-    range_records= mi_records_in_range(file,(int) i, &min_key, &max_key);
+    range_records= mi_records_in_range(file,(int) i, &min_key, &max_key,
+                                       &pages);
     if (range_records < info.records*8/10 ||
 	range_records > info.records*12/10)
     {
@@ -645,6 +647,7 @@ int main(int argc, char *argv[])
     if (j != 0 && k != 0)
     {
       key_range min_key, max_key;
+      page_range pages;
       if (j > k)
 	swap_variables(int, j, k);
       sprintf((char*) key,"%6d",j);
@@ -656,7 +659,7 @@ int main(int argc, char *argv[])
       max_key.key= key2;
       max_key.keypart_map= HA_WHOLE_KEY;
       max_key.flag= HA_READ_BEFORE_KEY;
-      range_records= mi_records_in_range(file, 0, &min_key, &max_key);
+      range_records= mi_records_in_range(file, 0, &min_key, &max_key, &pages);
       records=0;
       for (j++ ; j < k ; j++)
 	records+=key1[j];
diff --git a/storage/myisam/rt_test.c b/storage/myisam/rt_test.c
index 353e20951cd..11e824a8893 100644
--- a/storage/myisam/rt_test.c
+++ b/storage/myisam/rt_test.c
@@ -113,6 +113,7 @@ static int run_test(const char *filename)
   uchar read_record[MAX_REC_LENGTH];
   int upd= 10;
   ha_rows hrows;
+  page_range pages;
 
   bzero(&uniquedef, sizeof(uniquedef));
   bzero(&create_info, sizeof(create_info));
@@ -335,7 +336,7 @@ static int run_test(const char *filename)
   range.key= record+1;
   range.length= 1000;                           /* Big enough */
   range.flag= HA_READ_MBR_INTERSECT;
-  hrows= mi_records_in_range(file, 0, &range, (key_range*) 0);
+  hrows= mi_records_in_range(file, 0, &range, (key_range*) 0, &pages);
   printf("     %ld rows\n", (long) hrows);
 
   if (mi_close(file)) goto err;
diff --git a/storage/myisam/sp_test.c b/storage/myisam/sp_test.c
index 368c9dc2d5a..a33a6c8e743 100644
--- a/storage/myisam/sp_test.c
+++ b/storage/myisam/sp_test.c
@@ -70,6 +70,7 @@ int run_test(const char *filename)
   uchar read_record[MAX_REC_LENGTH];
   int upd=10;
   ha_rows hrows;
+  page_range pages;
   
   /* Define a column for NULLs and DEL markers*/
   
@@ -256,7 +257,7 @@ int run_test(const char *filename)
   max_range.key= record+1;
   max_range.length= 1000;                       /* Big enough */
   max_range.flag= HA_READ_KEY_EXACT;
-  hrows= mi_records_in_range(file, 0, &min_range, &max_range);
+  hrows= mi_records_in_range(file, 0, &min_range, &max_range, &pages);
   printf("     %ld rows\n", (long) hrows);
 
   if (mi_close(file)) goto err;
diff --git a/storage/myisammrg/ha_myisammrg.cc b/storage/myisammrg/ha_myisammrg.cc
index 4fd63020c6a..2a7ca8d51f8 100644
--- a/storage/myisammrg/ha_myisammrg.cc
+++ b/storage/myisammrg/ha_myisammrg.cc
@@ -1216,11 +1216,14 @@ void ha_myisammrg::position(const uchar *record)
 }
 
 
-ha_rows ha_myisammrg::records_in_range(uint inx, key_range *min_key,
-                                       key_range *max_key)
+ha_rows ha_myisammrg::records_in_range(uint inx,
+                                       const key_range *min_key,
+                                       const key_range *max_key,
+                                       page_range *pages)
 {
   DBUG_ASSERT(this->file->children_attached);
-  return (ha_rows) myrg_records_in_range(file, (int) inx, min_key, max_key);
+  return (ha_rows) myrg_records_in_range(file, (int) inx, min_key, max_key,
+                                         pages);
 }
 
 
diff --git a/storage/myisammrg/ha_myisammrg.h b/storage/myisammrg/ha_myisammrg.h
index b7cbd6c7d12..d5d62a002aa 100644
--- a/storage/myisammrg/ha_myisammrg.h
+++ b/storage/myisammrg/ha_myisammrg.h
@@ -129,7 +129,8 @@ public:
   int rnd_next(uchar *buf);
   int rnd_pos(uchar * buf, uchar *pos);
   void position(const uchar *record);
-  ha_rows records_in_range(uint inx, key_range *min_key, key_range *max_key);
+  ha_rows records_in_range(uint inx, const key_range *start_key,
+                           const key_range *end_key, page_range *pages);
   int delete_all_rows();
   int info(uint);
   int reset(void);
diff --git a/storage/myisammrg/myrg_range.c b/storage/myisammrg/myrg_range.c
index 893bda20833..da5e2c38d68 100644
--- a/storage/myisammrg/myrg_range.c
+++ b/storage/myisammrg/myrg_range.c
@@ -17,14 +17,21 @@
 #include "myrg_def.h"
 
 ha_rows myrg_records_in_range(MYRG_INFO *info, int inx,
-                              key_range *min_key, key_range *max_key)
+                              const key_range *min_key,
+                              const key_range *max_key,
+                              page_range *pages)
 {
   ha_rows records=0, res;
   MYRG_TABLE *table;
+  page_range ignore_pages;
+
+  /* Don't calculate pages of more than one active partition */
+  if (info->open_tables +1 != info->end_table)
+    pages= &ignore_pages;
 
   for (table=info->open_tables ; table != info->end_table ; table++)
   {
-    res= mi_records_in_range(table->table, inx, min_key, max_key);
+    res= mi_records_in_range(table->table, inx, min_key, max_key, pages);
     if (res == HA_POS_ERROR)
       return HA_POS_ERROR; 
     if (records > HA_POS_ERROR - res)
diff --git a/storage/oqgraph/ha_oqgraph.cc b/storage/oqgraph/ha_oqgraph.cc
index d84a59474f6..20a3732ce95 100644
--- a/storage/oqgraph/ha_oqgraph.cc
+++ b/storage/oqgraph/ha_oqgraph.cc
@@ -1185,8 +1185,10 @@ int ha_oqgraph::rename_table(const char *, const char *)
 }
 
 
-ha_rows ha_oqgraph::records_in_range(uint inx, key_range *min_key,
-                                  key_range *max_key)
+ha_rows ha_oqgraph::records_in_range(uint inx,
+                                     const key_range *min_key,
+                                     const key_range *max_key,
+                                     page_range *pages)
 {
   if (graph->get_thd() != current_thd) {
     DBUG_PRINT( "oq-debug", ("g->table->in_use: 0x%lx <-- current_thd 0x%lx", (long) graph->get_thd(), (long) current_thd));
diff --git a/storage/oqgraph/ha_oqgraph.h b/storage/oqgraph/ha_oqgraph.h
index 0c0af6def97..c8e175df616 100644
--- a/storage/oqgraph/ha_oqgraph.h
+++ b/storage/oqgraph/ha_oqgraph.h
@@ -99,7 +99,8 @@ public:
   int extra(enum ha_extra_function operation);
   int external_lock(THD *thd, int lock_type);
   int delete_all_rows(void);
-  ha_rows records_in_range(uint inx, key_range *min_key, key_range *max_key);
+  ha_rows records_in_range(uint inx, const key_range *min_key,
+                           const key_range *max_key, page_range *pages);
   int delete_table(const char *from);
   int rename_table(const char * from, const char * to);
   int create(const char *name, TABLE *form, HA_CREATE_INFO *create_info);
diff --git a/storage/rocksdb/ha_rocksdb.cc b/storage/rocksdb/ha_rocksdb.cc
index 3fa13295496..1b7ec9f9473 100644
--- a/storage/rocksdb/ha_rocksdb.cc
+++ b/storage/rocksdb/ha_rocksdb.cc
@@ -11910,8 +11910,9 @@ int ha_rocksdb::extra(enum ha_extra_function operation) {
   Given a starting key and an ending key, estimate the number of rows that
   will exist between the two keys.
 */
-ha_rows ha_rocksdb::records_in_range(uint inx, key_range *const min_key,
-                                     key_range *const max_key) {
+ha_rows ha_rocksdb::records_in_range(uint inx, const key_range *const min_key,
+                                     const key_range *const max_key,
+                                     page_range *pages) {
   DBUG_ENTER_FUNC();
 
   ha_rows ret = THDVAR(ha_thd(), records_in_range);
diff --git a/storage/rocksdb/ha_rocksdb.h b/storage/rocksdb/ha_rocksdb.h
index 426aa110597..1820326090f 100644
--- a/storage/rocksdb/ha_rocksdb.h
+++ b/storage/rocksdb/ha_rocksdb.h
@@ -902,8 +902,10 @@ public:
   int check(THD *const thd, HA_CHECK_OPT *const check_opt) override
       MY_ATTRIBUTE((__warn_unused_result__));
   int remove_rows(Rdb_tbl_def *const tbl);
-  ha_rows records_in_range(uint inx, key_range *const min_key,
-                           key_range *const max_key) override
+  ha_rows records_in_range(uint inx,
+                           const key_range *const min_key,
+                           const key_range *const max_key,
+                           page_range *pages) override
       MY_ATTRIBUTE((__warn_unused_result__));
 
   int delete_table(Rdb_tbl_def *const tbl);
diff --git a/storage/sequence/sequence.cc b/storage/sequence/sequence.cc
index 8dac7855a6e..8684a5c60b9 100644
--- a/storage/sequence/sequence.cc
+++ b/storage/sequence/sequence.cc
@@ -93,9 +93,8 @@ public:
   int index_prev(uchar *buf);
   int index_first(uchar *buf);
   int index_last(uchar *buf);
-  ha_rows records_in_range(uint inx, key_range *min_key,
-                                   key_range *max_key);
-
+  ha_rows records_in_range(uint inx, const key_range *start_key,
+                           const key_range *end_key, page_range *pages);
   double scan_time() { return (double)nvalues(); }
   double read_time(uint index, uint ranges, ha_rows rows) { return (double)rows; }
   double keyread_time(uint index, uint ranges, ha_rows rows) { return (double)rows; }
@@ -234,8 +233,9 @@ int ha_seq::index_last(uchar *buf)
   return index_prev(buf);
 }
 
-ha_rows ha_seq::records_in_range(uint inx, key_range *min_key,
-                                 key_range *max_key)
+ha_rows ha_seq::records_in_range(uint inx, const key_range *min_key,
+                                 const key_range *max_key,
+                                 page_range *pages)
 {
   ulonglong kmin= min_key ? uint8korr(min_key->key) : seqs->from;
   ulonglong kmax= max_key ? uint8korr(max_key->key) : seqs->to - 1;
diff --git a/storage/sphinx/ha_sphinx.cc b/storage/sphinx/ha_sphinx.cc
index df8ecd5f984..8403c767796 100644
--- a/storage/sphinx/ha_sphinx.cc
+++ b/storage/sphinx/ha_sphinx.cc
@@ -3360,7 +3360,7 @@ int ha_sphinx::rename_table ( const char *, const char * )
 // if start_key matches any rows.
 //
 // Called from opt_range.cc by check_quick_keys().
-ha_rows ha_sphinx::records_in_range ( uint, key_range *, key_range * )
+ha_rows ha_sphinx::records_in_range ( uint, const key_range *, const key_range *, page_range *)
 {
 	SPH_ENTER_METHOD();
 	SPH_RET(3); // low number to force index usage
diff --git a/storage/sphinx/ha_sphinx.h b/storage/sphinx/ha_sphinx.h
index decd88bad5a..cb46cb3dcbc 100644
--- a/storage/sphinx/ha_sphinx.h
+++ b/storage/sphinx/ha_sphinx.h
@@ -119,7 +119,7 @@ public:
 	int				reset();
 	int				external_lock ( THD * thd, int lock_type );
 	int				delete_all_rows ();
-	ha_rows			records_in_range ( uint inx, key_range * min_key, key_range * max_key );
+	ha_rows			        records_in_range ( uint inx, const key_range * min_key, const key_range * max_key,  page_range *pages);
 
 	int				delete_table ( const char * from );
 	int				rename_table ( const char * from, const char * to );
diff --git a/storage/spider/ha_spider.cc b/storage/spider/ha_spider.cc
index 694f612ff63..a7cfe45c8f7 100644
--- a/storage/spider/ha_spider.cc
+++ b/storage/spider/ha_spider.cc
@@ -8847,9 +8847,10 @@ int ha_spider::info(
 
 ha_rows ha_spider::records_in_range(
   uint inx,
-  key_range *start_key,
-  key_range *end_key
-) {
+  const key_range *start_key,
+  const key_range *end_key,
+  page_range *pages)
+{
   int error_num;
   THD *thd = ha_thd();
   double crd_interval = spider_param_crd_interval(thd, share->crd_interval);
diff --git a/storage/spider/ha_spider.h b/storage/spider/ha_spider.h
index edd7b8c881f..4a7b685527d 100644
--- a/storage/spider/ha_spider.h
+++ b/storage/spider/ha_spider.h
@@ -512,8 +512,9 @@ public:
   );
   ha_rows records_in_range(
     uint inx,
-    key_range *start_key,
-    key_range *end_key
+    const key_range *start_key,
+    const key_range *end_key,
+    page_range *pages
   );
   int check_crd();
   int pre_records();
diff --git a/storage/spider/spd_db_conn.cc b/storage/spider/spd_db_conn.cc
index e364c7b7f3a..2966f61c50e 100644
--- a/storage/spider/spd_db_conn.cc
+++ b/storage/spider/spd_db_conn.cc
@@ -5987,8 +5987,8 @@ int spider_db_show_index(
 }
 
 ha_rows spider_db_explain_select(
-  key_range *start_key,
-  key_range *end_key,
+  const key_range *start_key,
+  const key_range *end_key,
   ha_spider *spider,
   int link_idx
 ) {
diff --git a/storage/spider/spd_db_conn.h b/storage/spider/spd_db_conn.h
index 6fdb4b694ae..d4e527e4bf2 100644
--- a/storage/spider/spd_db_conn.h
+++ b/storage/spider/spd_db_conn.h
@@ -732,8 +732,8 @@ int spider_db_show_index(
 );
 
 ha_rows spider_db_explain_select(
-  key_range *start_key,
-  key_range *end_key,
+  const key_range *start_key,
+  const key_range *end_key,
   ha_spider *spider,
   int link_idx
 );
diff --git a/storage/spider/spd_db_handlersocket.cc b/storage/spider/spd_db_handlersocket.cc
index c5e663b42e9..9ffe59375c1 100644
--- a/storage/spider/spd_db_handlersocket.cc
+++ b/storage/spider/spd_db_handlersocket.cc
@@ -5421,8 +5421,8 @@ int spider_handlersocket_handler::append_delete_all_rows_part(
 }
 
 int spider_handlersocket_handler::append_explain_select_part(
-  key_range *start_key,
-  key_range *end_key,
+  const key_range *start_key,
+  const key_range *end_key,
   ulong sql_type,
   int link_idx
 ) {
diff --git a/storage/spider/spd_db_handlersocket.h b/storage/spider/spd_db_handlersocket.h
index 19a4a391ed6..de76e2c7163 100644
--- a/storage/spider/spd_db_handlersocket.h
+++ b/storage/spider/spd_db_handlersocket.h
@@ -843,8 +843,8 @@ public:
     ulong sql_type
   );
   int append_explain_select_part(
-    key_range *start_key,
-    key_range *end_key,
+    const key_range *start_key,
+    const key_range *end_key,
     ulong sql_type,
     int link_idx
   );
@@ -966,8 +966,8 @@ public:
     ulonglong &last_insert_id
   );
   ha_rows explain_select(
-    key_range *start_key,
-    key_range *end_key,
+    const key_range *start_key,
+    const key_range *end_key,
     int link_idx
   );
   int lock_tables(
diff --git a/storage/spider/spd_db_include.h b/storage/spider/spd_db_include.h
index 4daa5c71b93..8d4f972648a 100644
--- a/storage/spider/spd_db_include.h
+++ b/storage/spider/spd_db_include.h
@@ -1570,8 +1570,8 @@ public:
     ulong sql_type
   ) = 0;
   virtual int append_explain_select_part(
-    key_range *start_key,
-    key_range *end_key,
+    const key_range *start_key,
+    const key_range *end_key,
     ulong sql_type,
     int link_idx
   ) = 0;
@@ -1688,8 +1688,8 @@ public:
     ulonglong &last_insert_id
   ) = 0;
   virtual ha_rows explain_select(
-    key_range *start_key,
-    key_range *end_key,
+    const key_range *start_key,
+    const key_range *end_key,
     int link_idx
   ) = 0;
   virtual int lock_tables(
diff --git a/storage/spider/spd_db_mysql.cc b/storage/spider/spd_db_mysql.cc
index 155e814fae3..ca13ee38c7b 100644
--- a/storage/spider/spd_db_mysql.cc
+++ b/storage/spider/spd_db_mysql.cc
@@ -12690,8 +12690,8 @@ int spider_mbase_handler::append_truncate(
 }
 
 int spider_mbase_handler::append_explain_select_part(
-  key_range *start_key,
-  key_range *end_key,
+  const key_range *start_key,
+  const key_range *end_key,
   ulong sql_type,
   int link_idx
 ) {
@@ -12714,8 +12714,8 @@ int spider_mbase_handler::append_explain_select_part(
 
 int spider_mbase_handler::append_explain_select(
   spider_string *str,
-  key_range *start_key,
-  key_range *end_key,
+  const key_range *start_key,
+  const key_range *end_key,
   ulong sql_type,
   int link_idx
 ) {
@@ -14354,8 +14354,8 @@ int spider_mbase_handler::show_last_insert_id(
 }
 
 ha_rows spider_mbase_handler::explain_select(
-  key_range *start_key,
-  key_range *end_key,
+  const key_range *start_key,
+  const key_range *end_key,
   int link_idx
 ) {
   int error_num;
diff --git a/storage/spider/spd_db_mysql.h b/storage/spider/spd_db_mysql.h
index 626bb4d5624..547a98c8dff 100644
--- a/storage/spider/spd_db_mysql.h
+++ b/storage/spider/spd_db_mysql.h
@@ -1381,15 +1381,15 @@ public:
     int link_idx
   );
   int append_explain_select_part(
-    key_range *start_key,
-    key_range *end_key,
+    const key_range *start_key,
+    const key_range *end_key,
     ulong sql_type,
     int link_idx
   );
   int append_explain_select(
     spider_string *str,
-    key_range *start_key,
-    key_range *end_key,
+    const key_range *start_key,
+    const key_range *end_key,
     ulong sql_type,
     int link_idx
   );
@@ -1518,8 +1518,8 @@ public:
     ulonglong &last_insert_id
   );
   ha_rows explain_select(
-    key_range *start_key,
-    key_range *end_key,
+    const key_range *start_key,
+    const key_range *end_key,
     int link_idx
   );
   int lock_tables(
diff --git a/storage/spider/spd_db_oracle.cc b/storage/spider/spd_db_oracle.cc
index beaac6e8e24..dcdf6cb90d6 100644
--- a/storage/spider/spd_db_oracle.cc
+++ b/storage/spider/spd_db_oracle.cc
@@ -10150,8 +10150,8 @@ int spider_oracle_handler::append_truncate(
 }
 
 int spider_oracle_handler::append_explain_select_part(
-  key_range *start_key,
-  key_range *end_key,
+  const key_range *start_key,
+  const key_range *end_key,
   ulong sql_type,
   int link_idx
 ) {
@@ -10174,8 +10174,8 @@ int spider_oracle_handler::append_explain_select_part(
 
 int spider_oracle_handler::append_explain_select(
   spider_string *str,
-  key_range *start_key,
-  key_range *end_key,
+  const key_range *start_key,
+  const key_range *end_key,
   ulong sql_type,
   int link_idx
 ) {
@@ -11899,8 +11899,8 @@ int spider_oracle_handler::show_last_insert_id(
 }
 
 ha_rows spider_oracle_handler::explain_select(
-  key_range *start_key,
-  key_range *end_key,
+  const key_range *start_key,
+  const key_range *end_key,
   int link_idx
 ) {
   int error_num;
diff --git a/storage/spider/spd_db_oracle.h b/storage/spider/spd_db_oracle.h
index ebdc23a9bfa..fb00a2eebb1 100644
--- a/storage/spider/spd_db_oracle.h
+++ b/storage/spider/spd_db_oracle.h
@@ -1266,15 +1266,15 @@ public:
     int link_idx
   );
   int append_explain_select_part(
-    key_range *start_key,
-    key_range *end_key,
+    const key_range *start_key,
+    const key_range *end_key,
     ulong sql_type,
     int link_idx
   );
   int append_explain_select(
     spider_string *str,
-    key_range *start_key,
-    key_range *end_key,
+    const key_range *start_key,
+    const key_range *end_key,
     ulong sql_type,
     int link_idx
   );
@@ -1397,8 +1397,8 @@ public:
     ulonglong &last_insert_id
   );
   ha_rows explain_select(
-    key_range *start_key,
-    key_range *end_key,
+    const key_range *start_key,
+    const key_range *end_key,
     int link_idx
   );
   int lock_tables(
diff --git a/storage/tokudb/ha_tokudb.cc b/storage/tokudb/ha_tokudb.cc
index f4808a49c49..2c980196c1b 100644
--- a/storage/tokudb/ha_tokudb.cc
+++ b/storage/tokudb/ha_tokudb.cc
@@ -7889,7 +7889,9 @@ double ha_tokudb::index_only_read_time(uint keynr, double records) {
 //      number > 0 - There are approximately number matching rows in the range
 //      HA_POS_ERROR - Something is wrong with the index tree
 //
-ha_rows ha_tokudb::records_in_range(uint keynr, key_range* start_key, key_range* end_key) {
+ha_rows ha_tokudb::records_in_range(uint keynr, const key_range* start_key,
+                                    const key_range* end_key,
+                                    page_range *pages) {
     TOKUDB_HANDLER_DBUG_ENTER("%d %p %p", keynr, start_key, end_key);
     DBT *pleft_key, *pright_key;
     DBT left_key, right_key;
diff --git a/storage/tokudb/ha_tokudb.h b/storage/tokudb/ha_tokudb.h
index ea082aa53f4..5a87bf936fd 100644
--- a/storage/tokudb/ha_tokudb.h
+++ b/storage/tokudb/ha_tokudb.h
@@ -844,7 +844,9 @@ public:
     int external_lock(THD * thd, int lock_type);
     int start_stmt(THD * thd, thr_lock_type lock_type);
 
-    ha_rows records_in_range(uint inx, key_range * min_key, key_range * max_key);
+    ha_rows records_in_range(uint inx, const key_range * min_key,
+                             const key_range * max_key,
+                             page_range *pages);
 
     uint32_t get_cursor_isolation_flags(enum thr_lock_type lock_type, THD* thd);
     THR_LOCK_DATA **store_lock(THD * thd, THR_LOCK_DATA ** to, enum thr_lock_type lock_type);