MDEV-17573 Assertion in federatedx on multi-update

Cause: shared federatedx_io cannot store table-specific data.

Fix: move current row reference `federatedx_io_mysql::current` to
ha_federatedx.

FederatedX connection (represented by federatedx_io) is stored into
federatedx_txn::txn_list of per-server connections (see
federatedx_txn::acquire()). federatedx_txn object is stored into THD
(see ha_federatedx::external_lock()). When multiple handlers acquire
FederatedX connection they get single federatedx_io instance. Multiple
handlers do their operation via federatedx_io_mysql::mark_position()
and federatedx_io_mysql::fetch_row() in arbitrarty manner. They access
the same federatedx_io_mysql instance and same MYSQL_ROWS *current
pointer, so one handler disrupts the work of the other.

Related to "MDEV-14551 Can't find record in table on multi-table update
with ORDER BY".

4 files changed, 22 insertions, 15 deletions

diff --git a/storage/federatedx/federatedx_io_mysql.cc b/storage/federatedx/federatedx_io_mysql.cc
index 2068716eeba..4a2588a6af4 100644
--- a/storage/federatedx/federatedx_io_mysql.cc
+++ b/storage/federatedx/federatedx_io_mysql.cc
@@ -64,7 +64,6 @@ struct mysql_position
 class federatedx_io_mysql :public federatedx_io
 {
   MYSQL mysql; /* MySQL connection */
-  MYSQL_ROWS *current;
   DYNAMIC_ARRAY savepoints;
   bool requested_autocommit;
   bool actual_autocommit;
@@ -108,7 +107,8 @@ public:
   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);
-  virtual FEDERATEDX_IO_ROW *fetch_row(FEDERATEDX_IO_RESULT *io_result);
+  virtual FEDERATEDX_IO_ROW *fetch_row(FEDERATEDX_IO_RESULT *io_result,
+                                       FEDERATEDX_IO_ROWS **current= NULL);
   virtual ulong *fetch_lengths(FEDERATEDX_IO_RESULT *io_result);
   virtual const char *get_column_data(FEDERATEDX_IO_ROW *row,
                                       unsigned int column);
@@ -117,7 +117,7 @@ public:
 
   virtual size_t get_ref_length() const;
   virtual void mark_position(FEDERATEDX_IO_RESULT *io_result,
-                             void *ref);
+                             void *ref, FEDERATEDX_IO_ROWS *current);
   virtual int seek_position(FEDERATEDX_IO_RESULT **io_result,
                             const void *ref);
   virtual void set_thd(void *thd);
@@ -515,10 +515,12 @@ my_ulonglong federatedx_io_mysql::get_num_rows(FEDERATEDX_IO_RESULT *io_result)
 }
 
 
-FEDERATEDX_IO_ROW *federatedx_io_mysql::fetch_row(FEDERATEDX_IO_RESULT *io_result)
+FEDERATEDX_IO_ROW *federatedx_io_mysql::fetch_row(FEDERATEDX_IO_RESULT *io_result,
+                                                  FEDERATEDX_IO_ROWS **current)
 {
   MYSQL_RES *result= (MYSQL_RES*)io_result;
-  current= result->data_cursor;
+  if (current)
+    *current= (FEDERATEDX_IO_ROWS *) result->data_cursor;
   return (FEDERATEDX_IO_ROW *) mysql_fetch_row(result);
 }
 
@@ -626,11 +628,11 @@ size_t federatedx_io_mysql::get_ref_length() const
 
 
 void federatedx_io_mysql::mark_position(FEDERATEDX_IO_RESULT *io_result,
-                                        void *ref)
+                                        void *ref, FEDERATEDX_IO_ROWS *current)
 {
   mysql_position& pos= *reinterpret_cast<mysql_position*>(ref);
   pos.result= (MYSQL_RES *) io_result;
-  pos.offset= current;
+  pos.offset= (MYSQL_ROW_OFFSET) current;
 }
 
 int federatedx_io_mysql::seek_position(FEDERATEDX_IO_RESULT **io_result,
diff --git a/storage/federatedx/federatedx_io_null.cc b/storage/federatedx/federatedx_io_null.cc
index aa35d4bdecc..d945db528c7 100644
--- a/storage/federatedx/federatedx_io_null.cc
+++ b/storage/federatedx/federatedx_io_null.cc
@@ -90,7 +90,8 @@ public:
   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);
-  virtual FEDERATEDX_IO_ROW *fetch_row(FEDERATEDX_IO_RESULT *io_result);
+  virtual FEDERATEDX_IO_ROW *fetch_row(FEDERATEDX_IO_RESULT *io_result,
+                                       FEDERATEDX_IO_ROWS **current= NULL);
   virtual ulong *fetch_lengths(FEDERATEDX_IO_RESULT *io_result);
   virtual const char *get_column_data(FEDERATEDX_IO_ROW *row,
                                       unsigned int column);
@@ -98,7 +99,7 @@ public:
                               unsigned int column) const;
   virtual size_t get_ref_length() const;
   virtual void mark_position(FEDERATEDX_IO_RESULT *io_result,
-                             void *ref);
+                             void *ref, FEDERATEDX_IO_ROWS *current);
   virtual int seek_position(FEDERATEDX_IO_RESULT **io_result,
                             const void *ref);
 };
@@ -242,7 +243,8 @@ my_ulonglong federatedx_io_null::get_num_rows(FEDERATEDX_IO_RESULT *)
 }
 
 
-FEDERATEDX_IO_ROW *federatedx_io_null::fetch_row(FEDERATEDX_IO_RESULT *)
+FEDERATEDX_IO_ROW *federatedx_io_null::fetch_row(FEDERATEDX_IO_RESULT *,
+                                                 FEDERATEDX_IO_ROWS **current)
 {
   return NULL;
 }
@@ -288,7 +290,7 @@ size_t federatedx_io_null::get_ref_length() const
 
 
 void federatedx_io_null::mark_position(FEDERATEDX_IO_RESULT *io_result,
-                                       void *ref)
+                                       void *ref, FEDERATEDX_IO_ROWS *current)
 {
 }
 
diff --git a/storage/federatedx/ha_federatedx.cc b/storage/federatedx/ha_federatedx.cc
index 840a5a68885..5d2ddc1fb3b 100644
--- a/storage/federatedx/ha_federatedx.cc
+++ b/storage/federatedx/ha_federatedx.cc
@@ -2908,7 +2908,7 @@ int ha_federatedx::read_next(uchar *buf, FEDERATEDX_IO_RESULT *result)
     DBUG_RETURN(retval);
 
   /* Fetch a row, insert it back in a row format. */
-  if (!(row= io->fetch_row(result)))
+  if (!(row= io->fetch_row(result, &current)))
     DBUG_RETURN(HA_ERR_END_OF_FILE);
 
   if (!(retval= convert_row_to_internal_format(buf, row, result)))
@@ -2952,7 +2952,7 @@ void ha_federatedx::position(const uchar *record __attribute__ ((unused)))
   if (txn->acquire(share, ha_thd(), TRUE, &io))
     DBUG_VOID_RETURN;
 
-  io->mark_position(stored_result, ref);
+  io->mark_position(stored_result, ref, current);
 
   position_called= TRUE;
 
diff --git a/storage/federatedx/ha_federatedx.h b/storage/federatedx/ha_federatedx.h
index 56c34d9ef37..3d208d89074 100644
--- a/storage/federatedx/ha_federatedx.h
+++ b/storage/federatedx/ha_federatedx.h
@@ -129,6 +129,7 @@ typedef struct st_federatedx_share {
 
 typedef struct st_federatedx_result FEDERATEDX_IO_RESULT;
 typedef struct st_federatedx_row FEDERATEDX_IO_ROW;
+typedef struct st_federatedx_rows FEDERATEDX_IO_ROWS;
 typedef ptrdiff_t FEDERATEDX_IO_OFFSET;
 
 class federatedx_io
@@ -203,7 +204,8 @@ public:
   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;
-  virtual FEDERATEDX_IO_ROW *fetch_row(FEDERATEDX_IO_RESULT *io_result)=0;
+  virtual FEDERATEDX_IO_ROW *fetch_row(FEDERATEDX_IO_RESULT *io_result,
+                                       FEDERATEDX_IO_ROWS **current= NULL)=0;
   virtual ulong *fetch_lengths(FEDERATEDX_IO_RESULT *io_result)=0;
   virtual const char *get_column_data(FEDERATEDX_IO_ROW *row,
                                       unsigned int column)=0;
@@ -212,7 +214,7 @@ public:
 
   virtual size_t get_ref_length() const=0;
   virtual void mark_position(FEDERATEDX_IO_RESULT *io_result,
-                             void *ref)=0;
+                             void *ref, FEDERATEDX_IO_ROWS *current)=0;
   virtual int seek_position(FEDERATEDX_IO_RESULT **io_result,
                             const void *ref)=0;
   virtual void set_thd(void *thd) { }
@@ -265,6 +267,7 @@ class ha_federatedx: public handler
   federatedx_txn *txn;
   federatedx_io *io;
   FEDERATEDX_IO_RESULT *stored_result;
+  FEDERATEDX_IO_ROWS *current;
   /**
       Array of all stored results we get during a query execution.
   */