Merge lp:~capttofu/maria/bug_571200 into lp:maria/5.3

[CaptTofu]

revno: 2804
committer: Patrick Galbraith <patg@patg-desktop>
branch nick: maria-stock
timestamp: Fri 2010-07-23 15:37:32 -0400
message:
  Maria BUG# 571200, MySQL BUG# 32426: Fixed BLOG column order by for FederatedX

  Applied Antony's fixes to what I had started. These fixes were from his
  multi-bug fix at
  http://bazaar.launchpad.net/~atcurtis/maria/federatedx/revision/2833
  and are only the fixes for the ORDER BY BUG. I will also add separately the
  other fixes per-bug

  * Merged in fix for MySQL Bug#32426, http://lists.mysql.com/commits/102419
    However the io back-end is seperated from the front-end.
  * Added 3 new methods for the i/o class to abstract result set cursor
    positioning. Note to future self pertaining to IO subclasses for other
    drivers such as ODBC
  * Added test 'federated_bug_32426' to suite.

  https://bugs.launchpad.net/maria/+bug/571200

[Michael Widenius (monty)]

Hi!

>>>>> "CaptTofu" == CaptTofu <email address hidden> writes:

CaptTofu> CaptTofu has proposed merging lp:~capttofu/maria/bug_571200 into lp:maria.
CaptTofu> Requested reviews:
CaptTofu> Maria-captains (maria-captains)
CaptTofu> Related bugs:
CaptTofu> #571200 MySQL Bug#32426: FederatedX corrupt ORDER BY with TEXT
CaptTofu> https://bugs.launchpad.net/bugs/571200

Great!

I will review this tomorrow and if ok pull it into 5.1

Regards,
Monty

[Michael Widenius (monty)]

I was looking at merging this to 5.1 to fix the open federated bugs in 5.1, but the tree is depending on 5.2

i will look at this next week (after linuxconf) and see if I can easily extract the patch and apply it safely to 5.1

Unmerged revisions

2804. By Patrick Galbraith <patg@patg-desktop> on 2010-07-23

Maria BUG# 571200, MySQL BUG# 32426: Fixed BLOG column order by for FederatedX

Applied Antony's fixes to what I had started. These fixes were from his
multi-bug fix at
http://bazaar.launchpad.net/~atcurtis/maria/federatedx/revision/2833
and are only the fixes for the ORDER BY BUG. I will also add separately the
other fixes per-bug

* Merged in fix for MySQL Bug#32426, http://lists.mysql.com/commits/102419
  However the io back-end is seperated from the front-end.
* Added 3 new methods for the i/o class to abstract result set cursor
  positioning. Note to future self pertaining to IO subclasses for other
  drivers such as ODBC
* Added test 'federated_bug_32426' to suite.

https://bugs.launchpad.net/maria/+bug/571200

2803. By Patrick Galbraith <patg@patg-desktop> on 2010-07-21

Work in progress with federated_io methods

2802. By Patrick Galbraith <patg@patg-desktop> on 2010-07-20

Work in progress on adding federatedx_io methods

Methods for result set and result set positions

2801. By Patrick Galbraith <patg@patg-desktop> on 2010-07-19

Work in progress fixing bug 571200.

Most of patch from http://lists.mysql.com/commits/102419 has been
integrated into FederatedX. Need to figure out how to set/get the
results->data_cursor to work with FEDERATEDX_IO_RESULT struct.

=== added file 'mysql-test/suite/federated/federated_bug_32426.result'
--- mysql-test/suite/federated/federated_bug_32426.result	1970-01-01 00:00:00 +0000
+++ mysql-test/suite/federated/federated_bug_32426.result	2010-08-03 05:44:49 +0000
@@ -0,0 +1,30 @@
+CREATE DATABASE federated;
+CREATE DATABASE federated;
+#
+# Bug #32426: FEDERATED query returns corrupt results for ORDER BY 
+# on a TEXT column
+#
+CREATE TABLE federated.t1(a TEXT);
+INSERT INTO federated.t1 VALUES('abc'), ('gh'), ('f'), ('ijk'), ('de');
+CREATE TABLE federated.t1(a TEXT) ENGINE=FEDERATED
+CONNECTION='mysql://root@127.0.0.1:SLAVE_PORT/federated/t1';
+SELECT * FROM federated.t1 ORDER BY A;
+a
+abc
+de
+f
+gh
+ijk
+SELECT * FROM federated.t1 ORDER BY A DESC;
+a
+ijk
+gh
+f
+de
+abc
+DROP TABLE federated.t1;
+DROP TABLE federated.t1;
+DROP TABLE IF EXISTS federated.t1;
+DROP DATABASE IF EXISTS federated;
+DROP TABLE IF EXISTS federated.t1;
+DROP DATABASE IF EXISTS federated;

=== added file 'mysql-test/suite/federated/federated_bug_32426.test'
--- mysql-test/suite/federated/federated_bug_32426.test	1970-01-01 00:00:00 +0000
+++ mysql-test/suite/federated/federated_bug_32426.test	2010-08-03 05:44:49 +0000
@@ -0,0 +1,24 @@
+source federated.inc;
+
+--echo #
+--echo # Bug #32426: FEDERATED query returns corrupt results for ORDER BY 
+--echo # on a TEXT column
+--echo #
+connection slave;
+CREATE TABLE federated.t1(a TEXT);
+INSERT INTO federated.t1 VALUES('abc'), ('gh'), ('f'), ('ijk'), ('de');
+
+connection master;
+--replace_result $SLAVE_MYPORT SLAVE_PORT
+eval CREATE TABLE federated.t1(a TEXT) ENGINE=FEDERATED
+  CONNECTION='mysql://root@127.0.0.1:$SLAVE_MYPORT/federated/t1';
+SELECT * FROM federated.t1 ORDER BY A;
+SELECT * FROM federated.t1 ORDER BY A DESC;
+DROP TABLE federated.t1;
+
+connection slave;
+DROP TABLE federated.t1;
+
+connection default;
+
+source federated_cleanup.inc;

=== modified file 'storage/federated/ha_federated.cc'
--- storage/federated/ha_federated.cc	2010-06-01 19:52:20 +0000
+++ storage/federated/ha_federated.cc	2010-08-03 05:44:49 +0000
@@ -941,7 +941,7 @@
 static bool emit_key_part_name(String *to, KEY_PART_INFO *part)
 {
   DBUG_ENTER("emit_key_part_name");
-  if (append_ident(to, part->field->field_name, 
+  if (append_ident(to, part->field->field_name,
                    strlen(part->field->field_name), ident_quote_char))
     DBUG_RETURN(1);                           // Out of memory
   DBUG_RETURN(0);

=== modified file 'storage/federatedx/federatedx_io_mysql.cc'
--- storage/federatedx/federatedx_io_mysql.cc	2009-11-14 19:33:59 +0000
+++ storage/federatedx/federatedx_io_mysql.cc	2010-08-03 05:44:49 +0000
@@ -1,4 +1,4 @@
-/* 
+/*
 Copyright (c) 2007, Antony T Curtis
 All rights reserved.
 
@@ -51,6 +51,12 @@
   uint  flags;
 } SAVEPT;
 
+struct mysql_position
+{
+  MYSQL_RES* result;
+  MYSQL_ROW_OFFSET offset;
+};
+
 
 class federatedx_io_mysql :public federatedx_io
 {
@@ -76,16 +82,16 @@
 
   virtual int error_code();
   virtual const char *error_str();
-  
+
   void reset();
   int commit();
   int rollback();
-  
+
   int savepoint_set(ulong sp);
   ulong savepoint_release(ulong sp);
   ulong savepoint_rollback(ulong sp);
   void savepoint_restrict(ulong sp);
-  
+
   ulong last_savepoint() const;
   ulong actual_savepoint() const;
   bool is_autocommit() const;
@@ -94,7 +100,7 @@
                       uint table_name_length, uint flag);
 
   /* resultset operations */
-  
+
   virtual void free_result(FEDERATEDX_IO_RESULT *io_result);
   virtual unsigned int get_num_fields(FEDERATEDX_IO_RESULT *io_result);
   virtual my_ulonglong get_num_rows(FEDERATEDX_IO_RESULT *io_result);
@@ -104,6 +110,12 @@
                                       unsigned int column);
   virtual bool is_column_null(const FEDERATEDX_IO_ROW *row,
                               unsigned int column) const;
+
+  virtual size_t get_ref_length() const;
+  virtual void mark_position(FEDERATEDX_IO_RESULT *io_result,
+                             void *ref);
+  virtual int seek_position(FEDERATEDX_IO_RESULT **io_result,
+                            const void *ref);
 };
 
 
@@ -466,14 +478,13 @@
   return mysql_error(&mysql);
 }
 
-
 FEDERATEDX_IO_RESULT *federatedx_io_mysql::store_result()
 {
   FEDERATEDX_IO_RESULT *result;
   DBUG_ENTER("federatedx_io_mysql::store_result");
-  
+
   result= (FEDERATEDX_IO_RESULT *) mysql_store_result(&mysql);
-  
+
   DBUG_RETURN(result);
 }
 
@@ -590,3 +601,44 @@
   free_result(result);
   return 1;
 }
+
+
+
+size_t federatedx_io_mysql::get_ref_length() const
+{
+  return sizeof(mysql_position);
+}
+
+
+void federatedx_io_mysql::mark_position(FEDERATEDX_IO_RESULT *io_result,
+                                        void *ref)
+{
+  MYSQL_ROWS *tmp= 0;
+  mysql_position& pos= *reinterpret_cast<mysql_position*>(ref);
+  pos.result= (MYSQL_RES *) io_result;
+
+  if (pos.result && pos.result->data)
+  {
+    for (tmp= pos.result->data->data;
+         tmp && (tmp->next != pos.result->data_cursor);
+         tmp= tmp->next) ;
+  }
+
+  pos.offset= tmp;
+}
+
+int federatedx_io_mysql::seek_position(FEDERATEDX_IO_RESULT **io_result,
+                                       const void *ref)
+{
+  const mysql_position& pos= *reinterpret_cast<const mysql_position*>(ref);
+
+  if (!pos.result || !pos.offset)
+    return HA_ERR_END_OF_FILE;
+
+  pos.result->current_row= 0;
+  pos.result->data_cursor= pos.offset;
+  *io_result= (FEDERATEDX_IO_RESULT*) pos.result;
+
+  return 0;
+}
+

=== modified file 'storage/federatedx/federatedx_io_null.cc'
--- storage/federatedx/federatedx_io_null.cc	2009-11-14 19:33:59 +0000
+++ storage/federatedx/federatedx_io_null.cc	2010-08-03 05:44:49 +0000
@@ -96,6 +96,11 @@
                                       unsigned int column);
   virtual bool is_column_null(const FEDERATEDX_IO_ROW *row,
                               unsigned int column) const;
+  virtual size_t get_ref_length() const;
+  virtual void mark_position(FEDERATEDX_IO_RESULT *io_result,
+                             void *ref);
+  virtual int seek_position(FEDERATEDX_IO_RESULT **io_result,
+                            const void *ref);
 };
 
 
@@ -275,3 +280,20 @@
 
   return 0;
 }
+
+size_t federatedx_io_null::get_ref_length() const
+{
+  return sizeof(int);
+}
+
+
+void federatedx_io_null::mark_position(FEDERATEDX_IO_RESULT *io_result,
+                                       void *ref)
+{
+}
+
+int federatedx_io_null::seek_position(FEDERATEDX_IO_RESULT **io_result,
+                                      const void *ref)
+{
+  return 0;
+}

=== modified file 'storage/federatedx/ha_federatedx.cc'
--- storage/federatedx/ha_federatedx.cc	2010-04-01 14:34:51 +0000
+++ storage/federatedx/ha_federatedx.cc	2010-08-03 05:44:49 +0000
@@ -1717,14 +1717,14 @@
   return *txnp;
 }
 
-  
+
 int ha_federatedx::disconnect(handlerton *hton, MYSQL_THD thd)
 {
   federatedx_txn *txn= (federatedx_txn *) thd_get_ha_data(thd, hton);
   delete txn;
   return 0;
 }
- 
+
 
 /*
   Used for opening tables. The name will be the name of the file.
@@ -1756,14 +1756,15 @@
     free_share(txn, share);
     DBUG_RETURN(error);
   }
- 
+
+  ref_length= io->get_ref_length();
+
   txn->release(&io);
- 
-  ref_length= (table->s->primary_key != MAX_KEY ?
-               table->key_info[table->s->primary_key].key_length :
-               table->s->reclength);
+
   DBUG_PRINT("info", ("ref_length: %u", ref_length));
 
+  my_init_dynamic_array(&results, sizeof(FEDERATEDX_IO_RESULT*), 4, 4);
+
   reset();
 
   DBUG_RETURN(0);
@@ -1788,8 +1789,7 @@
   DBUG_ENTER("ha_federatedx::close");
 
   /* free the result set */
-  if (stored_result)
-    retval= free_result();
+  reset();
 
   /* Disconnect from mysql */
   if (!thd || !(txn= get_txn(thd, true)))
@@ -1799,7 +1799,7 @@
     tmp_txn.release(&io);
 
     DBUG_ASSERT(io == NULL);
-    
+
     if ((error= free_share(&tmp_txn, share)))
       retval= error;
   }
@@ -2525,7 +2525,7 @@
                                  uint key_len, enum ha_rkey_function find_flag)
 {
   int retval;
-  FEDERATEDX_IO_RESULT *io_result;
+  FEDERATEDX_IO_RESULT *io_result= 0;
   DBUG_ENTER("ha_federatedx::index_read_idx");
 
   if ((retval= index_read_idx_with_result_set(buf, index, key,
@@ -2601,7 +2601,7 @@
   if (!(retval= read_next(buf, *result)))
     DBUG_RETURN(retval);
 
-  io->free_result(*result);
+  insert_dynamic(&results, (uchar*) result);
   *result= 0;
   table->status= STATUS_NOT_FOUND;
   DBUG_RETURN(retval);
@@ -2669,10 +2669,7 @@
     DBUG_RETURN(retval);
 
   if (stored_result)
-  {
-    io->free_result(stored_result);
-    stored_result= 0;
-  }
+    (void) free_result();
 
   if (io->query(sql_query.ptr(), sql_query.length()))
   {
@@ -2773,10 +2770,7 @@
       DBUG_RETURN(error);
 
     if (stored_result)
-    {
-      io->free_result(stored_result);
-      stored_result= 0;
-    }
+      (void) free_result();
 
     if (io->query(share->select_query,
                   strlen(share->select_query)))
@@ -2803,17 +2797,35 @@
 int ha_federatedx::free_result()
 {
   int error;
-  federatedx_io *tmp_io= 0, **iop;
+  DBUG_ENTER("ha_federatedx::free_result");
   DBUG_ASSERT(stored_result);
-  if (!*(iop= &io) && (error= txn->acquire(share, TRUE, (iop= &tmp_io))))
-  {
-    DBUG_ASSERT(0);                             // Fail when testing
-    return error;
-  }
-  (*iop)->free_result(stored_result);
+  for (uint i= 0; i < results.elements; ++i)
+  {
+    FEDERATEDX_IO_RESULT *result= 0;
+    get_dynamic(&results, (uchar*) &result, i);
+    if (result == stored_result)
+      goto end;
+  }
+  if (position_called)
+  {
+    insert_dynamic(&results, (uchar*) &stored_result);
+  }
+  else
+  {
+    federatedx_io *tmp_io= 0, **iop;
+    if (!*(iop= &io) && (error= txn->acquire(share, TRUE, (iop= &tmp_io))))
+    {
+      DBUG_ASSERT(0);                             // Fail when testing
+      insert_dynamic(&results, (uchar*) &stored_result);
+      goto end;
+    }
+    (*iop)->free_result(stored_result);
+    txn->release(&tmp_io);
+  }
+end:
   stored_result= 0;
-  txn->release(&tmp_io);
-  return 0;
+  position_called= FALSE;
+  DBUG_RETURN(0);
 }
 
 int ha_federatedx::index_end(void)
@@ -2862,8 +2874,8 @@
 
   SYNOPSIS
     field_in_record_is_null()
-      buf       byte pointer to record 
-      result    mysql result set 
+      buf       byte pointer to record
+      result    mysql result set
 
     DESCRIPTION
      This method is a wrapper method that reads one record from a result
@@ -2896,24 +2908,43 @@
 }
 
 
-/*
-  store reference to current row so that we can later find it for
-  a re-read, update or delete.
-
-  In case of federatedx, a reference is either a primary key or
-  the whole record.
-
-  Called from filesort.cc, sql_select.cc, sql_delete.cc and sql_update.cc.
+/**
+  @brief      Store a reference to current row.
+
+  @details    During a query execution we may have different result sets (RS),
+              e.g. for different ranges. All the RS's used are stored in
+              memory and placed in @c results dynamic array. At the end of
+              execution all stored RS's are freed at once in the
+              @c ha_federated::reset().
+              So, in case of federated, a reference to current row is a
+              stored result address and current data cursor position.
+              As we keep all RS in memory during a query execution,
+              we can get any record using the reference any time until
+              @c ha_federated::reset() is called.
+              TODO: we don't have to store all RS's rows but only those
+              we call @c ha_federated::position() for, so we can free memory
+              where we store other rows in the @c ha_federated::index_end().
+
+  @param[in]  record  record data (unused)
+
 */
 
-void ha_federatedx::position(const uchar *record)
+void ha_federatedx::position(const uchar *record __attribute__ ((unused)))
 {
   DBUG_ENTER("ha_federatedx::position");
-  if (table->s->primary_key != MAX_KEY)
-    key_copy(ref, (uchar *)record, table->key_info + table->s->primary_key,
-             ref_length);
-  else
-    memcpy(ref, record, ref_length);
+
+  bzero(ref, ref_length);
+
+  if (!stored_result)
+    DBUG_VOID_RETURN;
+
+  if (txn->acquire(share, TRUE, &io))
+    DBUG_VOID_RETURN;
+
+  io->mark_position(stored_result, ref);
+
+  position_called= TRUE;
+
   DBUG_VOID_RETURN;
 }
 
@@ -2929,23 +2960,23 @@
 
 int ha_federatedx::rnd_pos(uchar *buf, uchar *pos)
 {
-  int result;
+  int retval;
+  FEDERATEDX_IO_RESULT *result= stored_result;
   DBUG_ENTER("ha_federatedx::rnd_pos");
   ha_statistic_increment(&SSV::ha_read_rnd_count);
-  if (table->s->primary_key != MAX_KEY)
-  {
-    /* We have a primary key, so use index_read_idx to find row */
-    result= index_read_idx(buf, table->s->primary_key, pos,
-                           ref_length, HA_READ_KEY_EXACT);
-  }
-  else
-  {
-    /* otherwise, get the old record ref as obtained in ::position */
-    memcpy(buf, pos, ref_length);
-    result= 0;
-  }
-  table->status= result ? STATUS_NOT_FOUND : 0;
-  DBUG_RETURN(result);
+
+  if ((retval= txn->acquire(share, TRUE, &io)))
+    goto error;
+
+  if ((retval= io->seek_position(&result, pos)))
+    goto error;
+
+  retval= read_next(buf, result);
+  DBUG_RETURN(retval);
+
+error:
+  table->status= STATUS_NOT_FOUND;
+  DBUG_RETURN(retval);
 }
 
 
@@ -2995,17 +3026,21 @@
 
 int ha_federatedx::info(uint flag)
 {
-  char error_buffer[FEDERATEDX_QUERY_BUFFER_SIZE];
   uint error_code;
+  THD *thd= current_thd;
+  federatedx_txn *tmp_txn;
   federatedx_io *tmp_io= 0, **iop= 0;
   DBUG_ENTER("ha_federatedx::info");
 
   error_code= ER_QUERY_ON_FOREIGN_DATA_SOURCE;
-  
+
+  // external_lock may not have been called so txn may not be set
+  tmp_txn= get_txn(thd);
+
   /* we want not to show table status if not needed to do so */
   if (flag & (HA_STATUS_VARIABLE | HA_STATUS_CONST | HA_STATUS_AUTO))
   {
-    if (!*(iop= &io) && (error_code= txn->acquire(share, TRUE, (iop= &tmp_io))))
+    if (!*(iop= &io) && (error_code= tmp_txn->acquire(share, TRUE, (iop= &tmp_io))))
       goto fail;
   }
 
@@ -3030,25 +3065,23 @@
     If ::info created it's own transaction, close it. This happens in case
     of show table status;
   */
-  txn->release(&tmp_io);
+  tmp_txn->release(&tmp_io);
 
   DBUG_RETURN(0);
 
 error:
   if (iop && *iop)
   {
-    my_sprintf(error_buffer, (error_buffer, ": %d : %s",
-                              (*iop)->error_code(), (*iop)->error_str()));
-    my_error(error_code, MYF(0), error_buffer);
+    my_printf_error((*iop)->error_code(), "Received error: %d : %s", MYF(0),
+                    (*iop)->error_code(), (*iop)->error_str());
   }
-  else
-  if (remote_error_number != -1 /* error already reported */)
+  else if (remote_error_number != -1 /* error already reported */)
   {
     error_code= remote_error_number;
     my_error(error_code, MYF(0), ER(error_code));
   }
 fail:
-  txn->release(&tmp_io);
+  tmp_txn->release(&tmp_io);
   DBUG_RETURN(error_code);
 }
 
@@ -3105,13 +3138,45 @@
 
 int ha_federatedx::reset(void)
 {
+  int error = 0;
+
   insert_dup_update= FALSE;
   ignore_duplicates= FALSE;
   replace_duplicates= FALSE;
-  return 0;
+  position_called= FALSE;
+
+  if (stored_result)
+    insert_dynamic(&results, (uchar*) &stored_result);
+  stored_result= 0;
+
+  if (results.elements)
+  {
+    federatedx_txn *tmp_txn;
+    federatedx_io *tmp_io= 0, **iop;
+
+    // external_lock may not have been called so txn may not be set
+    tmp_txn= get_txn(current_thd);
+
+    if (!*(iop= &io) && (error= tmp_txn->acquire(share, TRUE, (iop= &tmp_io))))
+    {
+      DBUG_ASSERT(0);                             // Fail when testing
+      return error;
+    }
+
+    for (uint i= 0; i < results.elements; ++i)
+    {
+      FEDERATEDX_IO_RESULT *result= 0;
+      get_dynamic(&results, (uchar*) &result, i);
+      (*iop)->free_result(result);
+    }
+    tmp_txn->release(&tmp_io);
+    reset_dynamic(&results);
+  }
+
+  return error;
+
 }
 
-
 /*
   Used to delete all rows in a table. Both for cases of truncate and
   for cases where the optimizer realizes that all rows will be
@@ -3237,7 +3302,7 @@
 
   str.length(0);
   str.append(STRING_WITH_LEN("SELECT * FROM "));
-  append_identifier(thd, &str, share->table_name, 
+  append_identifier(thd, &str, share->table_name,
                     share->table_name_length);
   str.append(STRING_WITH_LEN(" WHERE 1=0"));
 
@@ -3288,14 +3353,14 @@
   pthread_mutex_lock(&federatedx_mutex);
   tmp_share.s= get_server(&tmp_share, NULL);
   pthread_mutex_unlock(&federatedx_mutex);
-    
+
   if (tmp_share.s)
   {
     tmp_txn= get_txn(thd);
     if (!(retval= tmp_txn->acquire(&tmp_share, TRUE, &tmp_io)))
     {
       retval= test_connection(thd, tmp_io, &tmp_share);
-      tmp_txn->release(&tmp_io);    
+      tmp_txn->release(&tmp_io);
     }
     free_server(tmp_txn, tmp_share.s);
   }

=== modified file 'storage/federatedx/ha_federatedx.h'
--- storage/federatedx/ha_federatedx.h	2009-11-14 19:33:59 +0000
+++ storage/federatedx/ha_federatedx.h	2010-08-03 05:44:49 +0000
@@ -1,5 +1,5 @@
-/* 
-Copyright (c) 2008, Patrick Galbraith 
+/*
+Copyright (c) 2008, Patrick Galbraith
 All rights reserved.
 
 Redistribution and use in source and binary forms, with or without
@@ -43,7 +43,7 @@
 typedef struct st_fedrated_server {
   MEM_ROOT mem_root;
   uint use_count, io_count;
-  
+
   uchar *key;
   uint key_length;
 
@@ -74,10 +74,10 @@
 
 #include <mysql.h>
 
-/* 
+/*
   handler::print_error has a case statement for error numbers.
-  This value is (10000) is far out of range and will envoke the 
-  default: case.  
+  This value is (10000) is far out of range and will envoke the
+  default: case.
   (Current error range is 120-159 from include/my_base.h)
 */
 #define HA_FEDERATEDX_ERROR_WITH_REMOTE_SYSTEM 10000
@@ -158,7 +158,7 @@
   const char * get_database() const { return server->database; }
   ushort       get_port() const     { return server->port; }
   const char * get_socket() const   { return server->socket; }
-  
+
   static bool handles_scheme(const char *scheme);
   static federatedx_io *construct(MEM_ROOT *server_root,
                                   FEDERATEDX_SERVER *server);
@@ -167,7 +167,7 @@
   { return alloc_root(mem_root, size); }
   static void operator delete(void *ptr, size_t size)
   { TRASH(ptr, size); }
-    
+
   virtual int query(const char *buffer, uint length)=0;
   virtual FEDERATEDX_IO_RESULT *store_result()=0;
 
@@ -178,25 +178,25 @@
 
   virtual int error_code()=0;
   virtual const char *error_str()=0;
-  
+
   virtual void reset()=0;
   virtual int commit()=0;
   virtual int rollback()=0;
-  
+
   virtual int savepoint_set(ulong sp)=0;
   virtual ulong savepoint_release(ulong sp)=0;
   virtual ulong savepoint_rollback(ulong sp)=0;
   virtual void savepoint_restrict(ulong sp)=0;
-  
+
   virtual ulong last_savepoint() const=0;
   virtual ulong actual_savepoint() const=0;
   virtual bool is_autocommit() const=0;
 
   virtual bool table_metadata(ha_statistics *stats, const char *table_name,
                               uint table_name_length, uint flag) = 0;
-  
+
   /* resultset operations */
-  
+
   virtual void free_result(FEDERATEDX_IO_RESULT *io_result)=0;
   virtual unsigned int get_num_fields(FEDERATEDX_IO_RESULT *io_result)=0;
   virtual my_ulonglong get_num_rows(FEDERATEDX_IO_RESULT *io_result)=0;
@@ -206,6 +206,13 @@
                                       unsigned int column)=0;
   virtual bool is_column_null(const FEDERATEDX_IO_ROW *row,
                               unsigned int column) const=0;
+
+  virtual size_t get_ref_length() const=0;
+  virtual void mark_position(FEDERATEDX_IO_RESULT *io_result,
+                             void *ref)=0;
+  virtual int seek_position(FEDERATEDX_IO_RESULT **io_result,
+                            const void *ref)=0;
+
 };
 
 
@@ -215,12 +222,12 @@
   ulong savepoint_level;
   ulong savepoint_stmt;
   ulong savepoint_next;
-  
+
   void release_scan();
 public:
   federatedx_txn();
   ~federatedx_txn();
-  
+
   bool has_connections() const { return txn_list != NULL; }
   bool in_transaction() const { return savepoint_next != 0; }
   int acquire(FEDERATEDX_SHARE *share, bool readonly, federatedx_io **io);
@@ -254,8 +261,12 @@
   federatedx_txn *txn;
   federatedx_io *io;
   FEDERATEDX_IO_RESULT *stored_result;
+  /**
+      Array of all stored results we get during a query execution.
+  */
+  DYNAMIC_ARRAY results;
+  bool position_called;
   uint fetch_num; // stores the fetch num
-  FEDERATEDX_IO_OFFSET current_position;  // Current position used by ::position()
   int remote_error_number;
   char remote_error_buf[FEDERATEDX_QUERY_BUFFER_SIZE];
   bool ignore_duplicates, replace_duplicates;
@@ -269,7 +280,7 @@
   */
   uint convert_row_to_internal_format(uchar *buf, FEDERATEDX_IO_ROW *row,
                                       FEDERATEDX_IO_RESULT *result);
-  bool create_where_from_key(String *to, KEY *key_info, 
+  bool create_where_from_key(String *to, KEY *key_info,
                              const key_range *start_key,
                              const key_range *end_key,
                              bool records_in_range, bool eq_range);
@@ -348,18 +359,18 @@
     Talk to Kostja about this - how to get the
     number of rows * ...
     disk scan time on other side (block size, size of the row) + network time ...
-    The reason for "records * 1000" is that such a large number forces 
+    The reason for "records * 1000" is that such a large number forces
     this to use indexes "
   */
   double scan_time()
   {
     DBUG_PRINT("info", ("records %lu", (ulong) stats.records));
-    return (double)(stats.records*1000); 
+    return (double)(stats.records*1000);
   }
   /*
     The next method will never be called if you do not implement indexes.
   */
-  double read_time(uint index, uint ranges, ha_rows rows) 
+  double read_time(uint index, uint ranges, ha_rows rows)
   {
     /*
       Per Brian, this number is bugus, but this method must be implemented,