Fix for LP#571200 MySQL Bug#32426: FederatedX corrupt ORDER BY with TEXT

Patch taken from lp:~capttofu/maria/bug_571200 (originally for MariaDB 5.3, but adapted for 5.1)

mysql-test/suite/federated/federated_server.result

@@ -213,7 +213,7 @@ id	name
 alter server s1 options (database 'db_bogus');
 flush tables;
 select * from federated.t1;
-ERROR 42000: Got error: 1044 : Access denied for user 'test_fed'@'localhost' to database 'db_bogus'
+ERROR 42000: Received error: 1044 : Access denied for user 'test_fed'@'localhost' to database 'db_bogus'
 drop server if exists 's1';
 ERROR 42000: Access denied; you need the SUPER privilege for this operation
 create server 's1' foreign data wrapper 'mysql' options

storage/federatedx/federatedx_io_mysql.cc

@@ -51,6 +51,12 @@ typedef struct federatedx_savepoint
   uint  flags;
 } SAVEPT;
 
+struct mysql_position
+{
+  MYSQL_RES* result;
+  MYSQL_ROW_OFFSET offset;
+};
+
 
 class federatedx_io_mysql :public federatedx_io
 {
@@ -104,6 +110,12 @@ public:
                                       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,7 +478,6 @@ const char *federatedx_io_mysql::error_str()
   return mysql_error(&mysql);
 }
 
-
 FEDERATEDX_IO_RESULT *federatedx_io_mysql::store_result()
 {
   FEDERATEDX_IO_RESULT *result;
@@ -590,3 +601,45 @@ error:
   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;
+}
+

storage/federatedx/federatedx_io_null.cc

@@ -96,6 +96,11 @@ public:
                                       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 @@ bool federatedx_io_null::table_metadata(ha_statistics *stats,
 
   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;
+}

storage/federatedx/ha_federatedx.cc

@@ -1757,13 +1757,14 @@ int ha_federatedx::open(const char *name, int mode, uint test_if_locked)
     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 @@ int ha_federatedx::close(void)
   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)))
@@ -2525,7 +2525,7 @@ int ha_federatedx::index_read_idx(uchar *buf, uint index, const uchar *key,
                                  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 @@ int ha_federatedx::index_read_idx_with_result_set(uchar *buf, uint index,
   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 @@ int ha_federatedx::read_range_first(const key_range *start_key,
     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 @@ int ha_federatedx::rnd_init(bool scan)
       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::rnd_end()
 int ha_federatedx::free_result()
 {
   int error;
-  federatedx_io *tmp_io= 0, **iop;
+  DBUG_ENTER("ha_federatedx::free_result");
   DBUG_ASSERT(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
-    return error;
+      insert_dynamic(&results, (uchar*) &stored_result);
+      goto end;
     }
     (*iop)->free_result(stored_result);
-  stored_result= 0;
     txn->release(&tmp_io);
-  return 0;
+  }
+end:
+  stored_result= 0;
+  position_called= FALSE;
+  DBUG_RETURN(0);
 }
 
 int ha_federatedx::index_end(void)
@@ -2896,24 +2908,43 @@ int ha_federatedx::read_next(uchar *buf, FEDERATEDX_IO_RESULT *result)
 }
 
 
-/*
-  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.
+/**
+  @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)
 
-  Called from filesort.cc, sql_select.cc, sql_delete.cc and sql_update.cc.
 */
 
-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 @@ void ha_federatedx::position(const uchar *record)
 
 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);
 }
 
 
@@ -2996,15 +3027,20 @@ int ha_federatedx::rnd_pos(uchar *buf, uchar *pos)
 int ha_federatedx::info(uint flag)
 {
   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;
   }
 
@@ -3029,14 +3065,14 @@ int ha_federatedx::info(uint flag)
     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_printf_error((*iop)->error_code(), "Got error: %d : %s", MYF(0),
+    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 */)
@@ -3045,7 +3081,7 @@ error:
     my_error(error_code, MYF(0), ER(error_code));
   }
 fail:
-  txn->release(&tmp_io);
+  tmp_txn->release(&tmp_io);
   DBUG_RETURN(error_code);
 }
 
@@ -3105,12 +3141,44 @@ int ha_federatedx::extra(ha_extra_function operation)
 
 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

storage/federatedx/ha_federatedx.h

@@ -206,6 +206,13 @@ public:
                                       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;
+
 };
 
 
@@ -254,8 +261,12 @@ class ha_federatedx: public handler
   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;