MDEV-25080 Allow pushdown of UNIONs to foreign engines

Allow queries of multiple SELECTs combined together with
UNIONs/EXCEPTs/INTERSECTs to be pushed down to foreign engines.
If the foreign engine provides an interface method "create_unit"
and the UNIT is a top-level unit of the SQL query then the server
tries to push the whole SELECT_LEX_UNIT down to the engine for execution.
The engine should perform necessary checks and if they succeed,
execute the query. If the engine is unable to execute the whole unit,
then another attempt is made to push down SELECTs composing the unit
separately using the "create_select" interface method. In this case
the results of separate SELECTs are combined at the server side
thus composing the final result

mysql-test/suite/federated/federatedx_create_handlers.test

@@ -375,6 +375,267 @@ SELECT * FROM t10 LEFT JOIN
     WHERE id=2) dt2) dt
   ) ON t10.a=t11.a;
 
+--echo # MDEV-25080: Allow pushdown of queries involving UNIONs
+--echo #   in outer select to foreign engines
+--echo #
+
+connection master;
+
+TRUNCATE TABLE federated.t1;
+TRUNCATE TABLE federated.t2;
+
+INSERT INTO federated.t1 VALUES ('abc'), ('bcd'), ('cde');
+INSERT INTO federated.t2 VALUES ('abc'), ('bcd'), ('cde'), ('def'), ('efg');
+
+CREATE TABLE t3 (a varchar(30)) ENGINE=MyISAM;
+CREATE TABLE t4 (a varchar(30)) ENGINE=MyISAM;
+
+INSERT INTO t3 VALUES ('t3_myisam1'), ('t3_myisam2'), ('t3_myisam3');
+INSERT INTO t4 VALUES ('t4_myisam1'), ('t4_myisam2'), ('t4_myisam3');
+
+--echo # Pushdown of the whole UNION
+SELECT * from federated.t1 UNION SELECT * from federated.t2;
+EXPLAIN SELECT * from federated.t1 UNION SELECT * from federated.t2;
+
+--echo # Pushdown of a part of the UNION
+SELECT * from federated.t1 UNION SELECT * from t3;
+EXPLAIN SELECT * from federated.t1 UNION SELECT * from t3;
+
+SELECT * from federated.t1 UNION ALL SELECT * from federated.t2;
+EXPLAIN SELECT * from federated.t1 UNION ALL SELECT * from federated.t2;
+
+EXPLAIN FORMAT=JSON SELECT * from federated.t1 UNION ALL
+  SELECT * from federated.t2;
+
+ANALYZE SELECT * from federated.t1 UNION ALL SELECT * from federated.t2;
+
+--source include/analyze-format.inc
+ANALYZE FORMAT=JSON SELECT * from federated.t1 UNION ALL
+  SELECT * from federated.t2;
+
+SELECT * from federated.t1 EXCEPT SELECT * from federated.t2;
+
+EXPLAIN EXTENDED SELECT * from federated.t1 EXCEPT
+  SELECT * from federated.t2;
+
+EXPLAIN FORMAT=JSON SELECT * from federated.t1 EXCEPT
+  SELECT * from federated.t2;
+
+SELECT * from federated.t1 INTERSECT SELECT * from federated.t2;
+
+EXPLAIN PARTITIONS SELECT * from federated.t1 INTERSECT
+  SELECT * from federated.t2;
+
+EXPLAIN FORMAT=JSON SELECT * from federated.t1 INTERSECT
+  SELECT * from federated.t2;
+
+--echo # More than two SELECTs in a UNIT:
+SELECT * from federated.t1 INTERSECT
+  SELECT * from federated.t2 UNION ALL
+  SELECT * from federated.t2 EXCEPT
+  SELECT * from federated.t1;
+
+EXPLAIN
+  SELECT count(*) from federated.t1 INTERSECT
+  SELECT count(*) from federated.t2 UNION ALL
+  SELECT count(*)+20 from federated.t2 EXCEPT
+  SELECT count(*)+5 from federated.t1;
+  
+EXPLAIN FORMAT=JSON
+  SELECT count(*) from federated.t1 INTERSECT
+  SELECT count(*) from federated.t2 UNION ALL
+  SELECT count(*)+20 from federated.t2 EXCEPT
+  SELECT count(*)+5 from federated.t1;
+  
+ANALYZE
+  SELECT count(*) from federated.t1 INTERSECT
+  SELECT count(*) from federated.t2 UNION
+  SELECT count(*)+20 from federated.t2 EXCEPT
+  SELECT count(*)+5 from federated.t1;
+
+--echo # UNION inside a derived table: the whole derived table must be pushed
+SELECT * FROM 
+  (SELECT * FROM federated.t1 UNION ALL SELECT * FROM federated.t2) q;
+
+EXPLAIN 
+  SELECT * FROM 
+    (SELECT a FROM federated.t1 UNION ALL SELECT * FROM federated.t2) q;
+
+--echo # There is an uncacheable side effect due to fetch into @var,
+--echo # so the UNION cannot be pushed down as a whole.
+--echo # But separate SELECTs can be pushed, and the results are combined
+--echo # at the server side
+
+--disable_warnings
+
+SELECT count(*) FROM federated.t1 UNION
+  SELECT count(*) FROM federated.t1 EXCEPT
+  SELECT count(*)+1 FROM federated.t1
+  INTO @var;
+
+EXPLAIN SELECT count(*) FROM federated.t1 UNION
+  SELECT count(*) FROM federated.t2 EXCEPT
+  SELECT count(*)+1 FROM federated.t1
+  INTO @var;
+
+EXPLAIN FORMAT=JSON SELECT count(*) FROM federated.t1 UNION
+  SELECT count(*) FROM federated.t2 EXCEPT
+  SELECT count(*)+2 FROM federated.t2
+  INTO @var;
+
+--enable_warnings
+
+--echo # Prepared statements
+PREPARE stmt FROM "SELECT * from federated.t1 INTERSECT
+  SELECT * from federated.t2 UNION ALL
+  SELECT * from federated.t2 EXCEPT
+  SELECT * from federated.t1";
+
+EXECUTE stmt;
+EXECUTE stmt;
+EXECUTE stmt;
+
+PREPARE stmt FROM "EXPLAIN SELECT * from federated.t1 INTERSECT
+  SELECT * from federated.t2 UNION ALL
+  SELECT * from federated.t2 EXCEPT
+  SELECT * from federated.t1";
+
+EXECUTE stmt;
+EXECUTE stmt;
+
+--echo # UNIONs of mixed Federated/MyISAM tables, pushing parts of UNIONs
+SELECT * FROM federated.t1 UNION SELECT * FROM t3;
+EXPLAIN SELECT * FROM federated.t1 UNION SELECT * FROM t3;
+
+SELECT * FROM federated.t1 UNION ALL
+  SELECT * FROM t3 EXCEPT
+  SELECT * FROM federated.t2;
+EXPLAIN SELECT * FROM federated.t1 UNION ALL
+  SELECT * FROM t3 EXCEPT
+  SELECT * FROM federated.t2;
+
+SELECT * FROM t3 UNION ALL
+  SELECT * FROM federated.t1 EXCEPT
+  SELECT * FROM t4 INTERSECT
+  SELECT * FROM federated.t2;
+EXPLAIN SELECT * FROM t3 UNION ALL
+  SELECT * FROM federated.t1 EXCEPT
+  SELECT * FROM t4 INTERSECT
+  SELECT * FROM federated.t2;
+
+SELECT * FROM federated.t2 UNION ALL
+  SELECT * FROM t3 EXCEPT
+  SELECT * FROM t4 INTERSECT
+  SELECT * FROM federated.t1;
+EXPLAIN SELECT * FROM federated.t2 UNION ALL
+  SELECT * FROM t3 EXCEPT
+  SELECT * FROM t4 INTERSECT
+  SELECT * FROM federated.t1;
+
+--echo # Parenthesis must not prevent the whole UNIONs pushdown
+EXPLAIN (SELECT * FROM federated.t1 UNION
+  SELECT * FROM federated.t2) UNION ALL
+  SELECT * FROM federated.t1;
+
+(SELECT * FROM federated.t1 UNION
+  SELECT * FROM federated.t2) UNION ALL
+  SELECT * FROM federated.t1;
+
+EXPLAIN (SELECT * FROM federated.t1 UNION SELECT * FROM federated.t2)
+  UNION ALL (SELECT * FROM federated.t1 UNION SELECT * FROM federated.t2);
+
+(SELECT * FROM federated.t1 UNION SELECT * FROM federated.t2) UNION ALL
+  (SELECT * FROM federated.t1 UNION SELECT * FROM federated.t2);
+  
+--echo # Union of tables containing different INT data types
+connection slave;
+CREATE TABLE federated.t11 (a smallint(6) NOT NULL);
+INSERT INTO federated.t11 VALUES (-32678), (-1), (0);
+
+CREATE TABLE federated.t12 (a int(10) UNSIGNED NOT NULL);
+INSERT INTO federated.t12 VALUES (0), (1), (32767);
+
+connection master;
+
+--replace_result $SLAVE_MYPORT SLAVE_PORT
+eval
+CREATE TABLE federated.t11 (a smallint(6) NOT NULL)
+ENGINE="FEDERATED"
+CONNECTION='mysql://root@127.0.0.1:$SLAVE_MYPORT/federated/t11';
+
+--replace_result $SLAVE_MYPORT SLAVE_PORT
+eval
+CREATE TABLE federated.t12 (a int(10) UNSIGNED NOT NULL)
+ENGINE="FEDERATED"
+CONNECTION='mysql://root@127.0.0.1:$SLAVE_MYPORT/federated/t12';
+
+--echo # Entire UNION pushdown
+SELECT a FROM federated.t12 UNION ALL SELECT a FROM federated.t11;
+EXPLAIN SELECT a FROM federated.t12 UNION ALL SELECT a FROM federated.t11;
+
+SELECT a FROM federated.t11 UNION SELECT a FROM federated.t12;
+
+--echo # Partial pushdown of SELECTs composing the UNION
+SELECT a FROM federated.t12 UNION SELECT a FROM federated.t11 UNION SELECT 123;
+EXPLAIN
+  SELECT a FROM federated.t12 UNION SELECT a FROM federated.t11
+  UNION SELECT 123;
+
+SELECT a FROM federated.t12 EXCEPT
+  SELECT 1 UNION ALL
+  SELECT a FROM federated.t11 EXCEPT
+  SELECT 0;
+
+--echo # Union of tables containing different string data types
+connection slave;
+CREATE TABLE federated.t13 (a CHAR(6));
+INSERT INTO federated.t13 VALUES ('t13abc'), ('t13xx'), ('common');
+
+CREATE TABLE federated.t14 (a VARCHAR(8));
+INSERT INTO federated.t14 VALUES ('t14abcde'), ('t14xyzzz'), ('common');
+
+connection master;
+
+--replace_result $SLAVE_MYPORT SLAVE_PORT
+eval
+CREATE TABLE federated.t13 (a CHAR(6))
+ENGINE="FEDERATED"
+CONNECTION='mysql://root@127.0.0.1:$SLAVE_MYPORT/federated/t13';
+
+--replace_result $SLAVE_MYPORT SLAVE_PORT
+eval
+CREATE TABLE federated.t14 (a VARCHAR(8))
+ENGINE="FEDERATED"
+CONNECTION='mysql://root@127.0.0.1:$SLAVE_MYPORT/federated/t14';
+
+SELECT * FROM federated.t13 UNION SELECT * FROM federated.t14;
+EXPLAIN SELECT * FROM federated.t13 UNION SELECT * FROM federated.t14;
+
+SELECT * FROM federated.t14 UNION ALL SELECT * FROM federated.t13;
+
+SELECT * FROM federated.t14 UNION
+  SELECT * FROM federated.t13 UNION
+  SELECT '123456789000';
+  
+EXPLAIN SELECT * FROM federated.t14 UNION
+  SELECT * FROM federated.t13 UNION
+  SELECT '123456789000';
+
+SELECT * FROM federated.t13 UNION
+  SELECT '123456789000' UNION
+  SELECT * FROM federated.t14;
+
+# Cleanup
+connection master;
+DROP TABLES federated.t1, federated.t2, t3, t4, federated.t11, federated.t12,
+  federated.t13, federated.t14;
+
+connection slave;
+DROP TABLES federated.t1, federated.t2, federated.t11, federated.t12,
+  federated.t13, federated.t14;
+
+connection default;
+
 set global federated_pushdown=0;
 
 source include/federated_cleanup.inc;

sql/handler.h

@@ -56,6 +56,7 @@ class Field_string;
 class Field_varstring;
 class Field_blob;
 class Column_definition;
+class select_result;
 
 // the following is for checking tables
 
@@ -1206,6 +1207,7 @@ class derived_handler;
 class select_handler;
 struct Query;
 typedef class st_select_lex SELECT_LEX;
+typedef class st_select_lex_unit SELECT_LEX_UNIT;
 typedef struct st_order ORDER;
 
 /*
@@ -1551,10 +1553,18 @@ struct handlerton
   derived_handler *(*create_derived)(THD *thd, TABLE_LIST *derived);
 
   /*
-    Create and return a select_handler if the storage engine can execute
-    the select statement 'select, otherwise return NULL
+    Create and return a select_handler for a single SELECT.
+    If the storage engine cannot execute the select statement, return NULL
   */
-  select_handler *(*create_select) (THD *thd, SELECT_LEX *select);
+  select_handler *(*create_select) (THD *thd, SELECT_LEX *select_lex,
+                                   SELECT_LEX_UNIT *select_lex_unit);
+
+  /*
+    Create and return a select_handler for a unit (i.e. multiple SELECTs
+    combined with UNION/EXCEPT/INTERSECT). If the storage engine cannot execute
+    the statement, return NULL
+  */
+  select_handler *(*create_unit)(THD *thd, SELECT_LEX_UNIT *select_unit);
    
    /*********************************************************************
      Table discovery API.

sql/mysqld.h

@@ -922,7 +922,10 @@ enum enum_query_type
 
   // The temporary tables used by the query might be freed by the time
   // this print() call is made.
-  QT_DONT_ACCESS_TMP_TABLES= (1 << 13)
+  QT_DONT_ACCESS_TMP_TABLES= (1 << 13),
+
+  // Print only the SELECT part, even for INSERT...SELECT
+  QT_SELECT_ONLY = (1 << 14)
 };
 
 

sql/select_handler.cc

@@ -17,6 +17,7 @@
 #include "mariadb.h"
 #include "sql_priv.h"
 #include "sql_select.h"
+#include "sql_cte.h"
 #include "select_handler.h"
 
 
@@ -36,11 +37,26 @@
 */
 
 
-select_handler::select_handler(THD *thd_arg, handlerton *ht_arg)
- : thd(thd_arg), ht(ht_arg), table(NULL),
+select_handler::select_handler(THD *thd_arg, handlerton *ht_arg,
+                               SELECT_LEX *sel_lex)
+  : select_lex(sel_lex), lex_unit(nullptr), table(nullptr),
+    thd(thd_arg), ht(ht_arg), result(sel_lex->join->result),
     is_analyze(thd_arg->lex->analyze_stmt)
 {}
 
+select_handler::select_handler(THD *thd_arg, handlerton *ht_arg,
+                               SELECT_LEX_UNIT *sel_unit)
+  : select_lex(nullptr), lex_unit(sel_unit), table(nullptr),
+    thd(thd_arg), ht(ht_arg), result(sel_unit->result),
+    is_analyze(thd_arg->lex->analyze_stmt)
+{}
+
+select_handler::select_handler(THD *thd_arg, handlerton *ht_arg,
+                               SELECT_LEX *sel_lex, SELECT_LEX_UNIT *sel_unit)
+    : select_lex(sel_lex), lex_unit(sel_unit), table(nullptr), thd(thd_arg),
+      ht(ht_arg), result(sel_lex->join->result),
+      is_analyze(thd_arg->lex->analyze_stmt)
+{}
 
 select_handler::~select_handler()
 {
@@ -49,18 +65,37 @@ select_handler::~select_handler()
 }
 
 
-TABLE *select_handler::create_tmp_table(THD *thd, SELECT_LEX *select)
+TABLE *select_handler::create_tmp_table(THD *thd)
 {
   List<Item> types;
   TMP_TABLE_PARAM tmp_table_param;
-  TABLE *table;
   DBUG_ENTER("select_handler::create_tmp_table");
 
-  if (select->master_unit()->join_union_item_types(thd, types, 1))
+  SELECT_LEX_UNIT *unit= nullptr;
+  uint unit_parts_count= 0;
+
+  if (lex_unit)
+  {
+    unit= lex_unit;
+    SELECT_LEX *sl= unit->first_select();
+    while (sl)
+    {
+      unit_parts_count++;
+      sl= sl->next_select();
+    }
+  }
+  else
+  {
+    unit= select_lex->master_unit();
+    unit_parts_count= 1;
+  }
+
+  if (unit->join_union_item_types(thd, types, unit_parts_count))
     DBUG_RETURN(NULL);
+
   tmp_table_param.init();
   tmp_table_param.field_count= tmp_table_param.func_count= types.elements;
-  table= ::create_tmp_table(thd, &tmp_table_param, types,
+  TABLE *table= ::create_tmp_table(thd, &tmp_table_param, types,
                                    (ORDER *) 0, false, 0,
                                    TMP_TABLE_ALL_COLUMNS, 1,
                                    &empty_clex_str, true, false);
@@ -75,7 +110,7 @@ bool select_handler::prepare()
     Some engines (e.g. XPand) initialize "table" on their own.
     So we need to create a temporary table only if "table" is NULL.
   */
-  if (!table && !(table= create_tmp_table(thd, select)))
+  if (!table && !(table= create_tmp_table(thd)))
     DBUG_RETURN(true);
   DBUG_RETURN(table->fill_item_list(&result_columns));
 }
@@ -92,22 +127,19 @@ bool select_handler::send_result_set_metadata()
     DBUG_RETURN(false);
   }
   #endif /* WITH_WSREP */
-  if (select->join->result->send_result_set_metadata(result_columns,
-                                         Protocol::SEND_NUM_ROWS |
-                                         Protocol::SEND_EOF))
-    DBUG_RETURN(true);
   
-  DBUG_RETURN(false);
+  DBUG_RETURN(result->send_result_set_metadata(
+        result_columns, Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF));
 }
 
 
 bool select_handler::send_data()
 {
-  DBUG_ENTER("Pushdown_select::send_data");
-
-  if (select->join->result->send_data(result_columns))
+  DBUG_ENTER("select_handler::send_data");
+  int res= result->send_data(result_columns);
+  // "-1" means "duplicate when executing UNION"
+  if (res && res != -1)
     DBUG_RETURN(true);
-
   DBUG_RETURN(false);
 }
 
@@ -115,10 +147,7 @@ bool select_handler::send_data()
 bool select_handler::send_eof()
 {
   DBUG_ENTER("select_handler::send_eof");
-
-  if (select->join->result->send_eof())
-    DBUG_RETURN(true);
-  DBUG_RETURN(false);
+  DBUG_RETURN(result->send_eof());
 }
 
 

sql/select_handler.h

@@ -30,25 +30,18 @@
 class select_handler
 {
  public:
-  THD *thd;
-  handlerton *ht;
+   // Constructor for a single SELECT_LEX (not a part of a unit)
+  select_handler(THD *thd_arg, handlerton *ht_arg, SELECT_LEX *sel_lex);
 
-  SELECT_LEX *select;  // Select to be excuted
+  // Constructor for a unit (UNION/EXCEPT/INTERSECT)
+  select_handler(THD *thd_arg, handlerton *ht_arg, SELECT_LEX_UNIT *sel_unit);
 
   /*
-    Temporary table where all results should be stored in record[0]
-    The table has a field for every item from the select_lex::item_list.
-    The table is actually never filled. Only its record buffer is used.
+    Constructor for a SELECT_LEX which is a part of a unit
+    (partial pushdown). Both SELECT_LEX and SELECT_LEX_UNIT are passed
   */
-  TABLE *table;
-  List<Item> result_columns;
-
-  bool is_analyze;
-
-  bool send_result_set_metadata();
-  bool send_data();
-
-  select_handler(THD *thd_arg, handlerton *ht_arg);
+  select_handler(THD *thd_arg, handlerton *ht_arg, SELECT_LEX *sel_lex,
+                 SELECT_LEX_UNIT *sel_unit);
 
   virtual ~select_handler();
 
@@ -56,9 +49,29 @@ class select_handler
 
   virtual bool prepare();
 
-  static TABLE *create_tmp_table(THD *thd, SELECT_LEX *sel);
+  /*
+    Select_handler processes one of
+    - single SELECT
+    - whole unit (multiple SELECTs combined with UNION/EXCEPT/INTERSECT)
+    - single SELECT that is part of a unit (partial pushdown)
+
+    In the case of single SELECT select_lex is initialized and lex_unit==NULL,
+    in the case of whole UNIT select_lex == NULL and lex_unit is initialized,
+    in the case of partial pushdown both select_lex and lex_unit
+      are initialized
+  */
+  SELECT_LEX *select_lex;      // Single select to be executed
+  SELECT_LEX_UNIT *lex_unit;   // Unit to be executed
+
+  /*
+    Temporary table where all results should be stored in record[0]
+    The table has a field for every item from the select_lex::item_list.
+    The table is actually never filled. Only its record buffer is used.
+  */
+  TABLE *table;
 
 protected:
+
   /*
     Functions to scan the select result set.
     All these returns 0 if ok, error code in case of error.
@@ -80,7 +93,19 @@ class select_handler
   /* Report errors */
   virtual void print_error(int error, myf errflag);
 
+  bool send_result_set_metadata();
+  bool send_data();
   bool send_eof();
+
+  TABLE *create_tmp_table(THD *thd);
+
+  THD *thd;
+  handlerton *ht;
+
+  select_result *result;        // Object receiving the retrieved data
+  List<Item> result_columns;
+
+  bool is_analyze;
 };
 
 #endif /* SELECT_HANDLER_INCLUDED */

sql/sql_derived.cc

@@ -894,22 +894,6 @@ bool mysql_derived_prepare(THD *thd, LEX *lex, TABLE_LIST *derived)
     first_select->mark_as_belong_to_derived(derived);
 
   derived->dt_handler= derived->find_derived_handler(thd);
-  if (derived->dt_handler)
-  {
-    char query_buff[4096];
-    String derived_query(query_buff, sizeof(query_buff), thd->charset());
-    derived_query.length(0);
-    derived->derived->print(&derived_query,
-                            enum_query_type(QT_VIEW_INTERNAL | 
-                                            QT_ITEM_ORIGINAL_FUNC_NULLIF |
-                                            QT_PARSABLE));
-    if (!thd->make_lex_string(&derived->derived_spec,
-                              derived_query.ptr(), derived_query.length()))
-    {
-      delete derived->dt_handler;
-      derived->dt_handler= NULL;
-    }
-  }
 
 exit:
   /* Hide "Unknown column" or "Unknown function" error */

sql/sql_explain.cc

@@ -36,6 +36,11 @@ const char *unit_operation_text[4]=
    "UNIT RESULT","UNION RESULT","INTERSECT RESULT","EXCEPT RESULT"
 };
 
+const char *pushed_unit_operation_text[4]=
+{
+  "PUSHED UNIT", "PUSHED UNION", "PUSHED INTERSECT", "PUSHED EXCEPT"
+};
+
 const char *pushed_derived_text= "PUSHED DERIVED";
 const char *pushed_select_text= "PUSHED SELECT";
 
@@ -593,6 +598,21 @@ uint Explain_union::make_union_table_name(char *buf)
 
 
 int Explain_union::print_explain(Explain_query *query,
+                                 select_result_sink *output,
+                                 uint8 explain_flags, bool is_analyze)
+{
+  if (is_pushed_down_to_engine)
+    return print_explain_pushed_down(output, explain_flags, is_analyze);
+  else
+    return print_explain_regular(query, output, explain_flags, is_analyze);
+}
+
+/*
+  Prints EXPLAIN plan for a regular UNIT (UNION/EXCEPT/INTERSECT),
+  i.e. UNIT that has not been pushed down to a storage engine
+*/
+
+int Explain_union::print_explain_regular(Explain_query *query,
                                  select_result_sink *output,
                                  uint8 explain_flags, 
                                  bool is_analyze)
@@ -609,7 +629,15 @@ int Explain_union::print_explain(Explain_query *query,
   }
 
   if (!using_tmp)
+  {
+    /*
+      The union operation may not employ a temporary table, for example,
+      for UNION ALL, in that case the results of the query are sent directly
+      to the output. So there is no actual UNION operation and we don't need
+      to print the line in the EXPLAIN output.
+    */
     return 0;
+  }
 
   /* Print a line with "UNIT RESULT" */
   List<Item> item_list;
@@ -682,7 +710,6 @@ int Explain_union::print_explain(Explain_query *query,
                                       extra_buf.length()),
                       mem_root);
 
-  //output->unit.offset_limit_cnt= 0; 
   if (output->send_data(item_list))
     return 1;
   
@@ -694,9 +721,89 @@ int Explain_union::print_explain(Explain_query *query,
 }
 
 
+/*
+  Prints EXPLAIN plan for a UNIT (UNION/EXCEPT/INTERSECT) that
+  has been pushed down to a storage engine
+*/
+
+int Explain_union::print_explain_pushed_down(select_result_sink *output,
+                                             uint8 explain_flags,
+                                             bool is_analyze)
+{
+  THD *thd= output->thd;
+  MEM_ROOT *mem_root= thd->mem_root;
+  List<Item> item_list;
+  Item *item_null= new (mem_root) Item_null(thd);
+
+  /* `id` column */
+  item_list.push_back(item_null, mem_root);
+
+  /* `select_type` column */
+  push_str(thd, &item_list, fake_select_type);
+
+  /* `table` column */
+  item_list.push_back(item_null, mem_root);
+
+  /* `partitions` column */
+  if (explain_flags & DESCRIBE_PARTITIONS)
+    item_list.push_back(item_null, mem_root);
+
+  /* `type` column */
+  item_list.push_back(item_null, mem_root);
+
+  /* `possible_keys` column */
+  item_list.push_back(item_null, mem_root);
+
+  /* `key` */
+  item_list.push_back(item_null, mem_root);
+
+  /* `key_len` */
+  item_list.push_back(item_null, mem_root);
+
+  /* `ref` */
+  item_list.push_back(item_null, mem_root);
+
+  /* `rows` */
+  item_list.push_back(item_null, mem_root);
+
+  /* `r_rows` */
+  if (is_analyze)
+    item_list.push_back(item_null, mem_root);
+
+  /* `filtered` */
+  if (explain_flags & DESCRIBE_EXTENDED || is_analyze)
+    item_list.push_back(item_null, mem_root);
+
+  /* `r_filtered` */
+  if (is_analyze)
+    item_list.push_back(item_null, mem_root);
+
+  /* `Extra` */
+  item_list.push_back(item_null, mem_root);
+
+  if (output->send_data(item_list))
+    return 1;
+  return 0;
+}
+
+
 void Explain_union::print_explain_json(Explain_query *query,
                                        Json_writer *writer, bool is_analyze,
                                        bool no_tmp_tbl)
+{
+  if (is_pushed_down_to_engine)
+    print_explain_json_pushed_down(query, writer, is_analyze, no_tmp_tbl);
+  else
+    print_explain_json_regular(query, writer, is_analyze, no_tmp_tbl);
+}
+
+/*
+  Prints EXPLAIN plan in JSON format for a regular UNIT (UNION/EXCEPT/INTERSECT),
+  i.e. UNIT that has not been pushed down to a storage engine
+*/
+
+void Explain_union::print_explain_json_regular(
+  Explain_query *query, Json_writer *writer, bool is_analyze, bool no_tmp_tbl)
 {
   Json_writer_nesting_guard guard(writer);
   char table_name_buffer[SAFE_NAME_LEN];
@@ -755,6 +862,31 @@ void Explain_union::print_explain_json(Explain_query *query,
     writer->end_object();
 }
 
+/*
+  Prints EXPLAIN plan in JSON format for a UNIT (UNION/EXCEPT/INTERSECT) that
+  has been pushed down to a storage engine
+*/
+
+void Explain_union::print_explain_json_pushed_down(Explain_query *query,
+                                                   Json_writer *writer,
+                                                   bool is_analyze,
+                                                   bool no_tmp_tbl)
+{
+  Json_writer_nesting_guard guard(writer);
+
+  writer->add_member("query_block").start_object();
+
+  if (is_recursive_cte)
+    writer->add_member("recursive_union").start_object();
+  else
+    writer->add_member("union_result").start_object();
+
+  writer->add_member("message").add_str(fake_select_type);
+
+  writer->end_object(); // union_result
+  writer->end_object(); // query_block
+}
+
 
 /*
   Print EXPLAINs for all children nodes (i.e. for subqueries)

sql/sql_explain.h

@@ -336,6 +336,7 @@ class Explain_aggr_window_funcs : public Explain_aggr_node
 /////////////////////////////////////////////////////////////////////////////
 
 extern const char *unit_operation_text[4];
+extern const char *pushed_unit_operation_text[4];
 extern const char *pushed_derived_text;
 extern const char *pushed_select_text;
 
@@ -350,7 +351,7 @@ class Explain_union : public Explain_node
 public:
   Explain_union(MEM_ROOT *root, bool is_analyze) : 
     Explain_node(root), union_members(PSI_INSTRUMENT_MEM),
-    is_recursive_cte(false),
+    is_recursive_cte(false), is_pushed_down_to_engine(false),
     fake_select_lex_explain(root, is_analyze)
   {}
 
@@ -383,11 +384,17 @@ class Explain_union : public Explain_node
                     uint8 explain_flags, bool is_analyze);
   void print_explain_json(Explain_query *query, Json_writer *writer,
                           bool is_analyze, bool no_tmp_tbl);
+  void print_explain_json_regular(Explain_query *query, Json_writer *writer,
+                          bool is_analyze, bool no_tmp_tbl);
+  void print_explain_json_pushed_down(Explain_query *query,
+                                      Json_writer *writer, bool is_analyze,
+                                      bool no_tmp_tbl);
 
   const char *fake_select_type;
   bool using_filesort;
   bool using_tmp;
   bool is_recursive_cte;
+  bool is_pushed_down_to_engine;
   
   /*
     Explain data structure for "fake_select_lex" (i.e. for the degenerate
@@ -406,6 +413,10 @@ class Explain_union : public Explain_node
   }
 private:
   uint make_union_table_name(char *buf);
+  int print_explain_regular(Explain_query *query, select_result_sink *output,
+                            uint8 explain_flags, bool is_analyze);
+  int print_explain_pushed_down(select_result_sink *output,
+                                uint8 explain_flags, bool is_analyze);
   
   Table_access_tracker fake_select_lex_tracker;
   /* This one is for reading after ORDER BY */

sql/sql_lex.cc

@@ -5450,11 +5450,17 @@ void st_select_lex::set_explain_type(bool on_the_fly)
       using_materialization= TRUE;
   }
 
-  if (master_unit()->thd->lex->first_select_lex() == this)
-  {
+  if (!on_the_fly)
+    options|= SELECT_DESCRIBE;
+
   if (pushdown_select)
+  {
     type= pushed_select_text;
-    else
+    return;
+  }
+
+  if (master_unit()->thd->lex->first_select_lex() == this)
+  {
     type= is_primary ? "PRIMARY" : "SIMPLE";
   }
   else
@@ -5480,10 +5486,7 @@ void st_select_lex::set_explain_type(bool on_the_fly)
         if (is_uncacheable & UNCACHEABLE_DEPENDENT)
           type= "DEPENDENT SUBQUERY";
         else
-         {
-           type= is_uncacheable? "UNCACHEABLE SUBQUERY" :
-                                 "SUBQUERY";
-         }
+          type= is_uncacheable ? "UNCACHEABLE SUBQUERY" : "SUBQUERY";
       }
     }
     else
@@ -5506,7 +5509,10 @@ void st_select_lex::set_explain_type(bool on_the_fly)
         {
           type= is_uncacheable ? "UNCACHEABLE UNION": "UNION";
           if (this == master_unit()->fake_select_lex)
-            type= unit_operation_text[master_unit()->common_op()];
+            type=
+                master_unit()->pushdown_unit
+                    ? pushed_unit_operation_text[master_unit()->common_op()]
+                    : unit_operation_text[master_unit()->common_op()];
           /*
             join below may be =NULL when this functions is called at an early
             stage. It will be later called again and we will set the correct
@@ -5541,9 +5547,6 @@ void st_select_lex::set_explain_type(bool on_the_fly)
       }
     }
   }
-
-  if (!on_the_fly)
-    options|= SELECT_DESCRIBE;
 }
 
 
@@ -5967,7 +5970,10 @@ int st_select_lex_unit::save_union_explain(Explain_query *output)
   for (SELECT_LEX *sl= first; sl; sl= sl->next_select())
     eu->add_select(sl->select_number);
 
-  eu->fake_select_type= unit_operation_text[eu->operation= common_op()];
+  eu->is_pushed_down_to_engine= (pushdown_unit != nullptr);
+  eu->fake_select_type= pushdown_unit ?
+    pushed_unit_operation_text[eu->operation= common_op()] :
+    unit_operation_text[eu->operation= common_op()];
   eu->using_filesort= MY_TEST(global_parameters()->order_list.first);
   eu->using_tmp= union_needs_tmp_table();
 

sql/sql_lex.h

@@ -38,6 +38,7 @@
 #include "sql_schema.h"
 #include "table.h"
 #include "sql_class.h"                // enum enum_column_usage
+#include "select_handler.h"
 
 /* Used for flags of nesting constructs */
 #define SELECT_NESTING_MAP_SIZE 64
@@ -872,7 +873,7 @@ class st_select_lex_unit: public st_select_lex_node {
   st_select_lex_unit()
     : union_result(NULL), table(NULL),  result(NULL), fake_select_lex(NULL),
       last_procedure(NULL),cleaned(false), bag_set_op_optimized(false),
-      have_except_all_or_intersect_all(false)
+      have_except_all_or_intersect_all(false), pushdown_unit(NULL)
   {
   }
 
@@ -937,6 +938,10 @@ class st_select_lex_unit: public st_select_lex_node {
   bool bag_set_op_optimized:1;
   bool optimize_started:1;
   bool have_except_all_or_intersect_all:1;
+
+  /* The object used to organize execution of the UNIT by a foreign engine */
+  select_handler *pushdown_unit;
+
   /**
      TRUE if the unit contained TVC at the top level that has been wrapped
      into SELECT:
@@ -1049,6 +1054,7 @@ class st_select_lex_unit: public st_select_lex_node {
   friend class st_select_lex;
 
 private:
+  bool exec_inner();
   bool is_derived_eliminated() const;
 };
 
@@ -1145,8 +1151,6 @@ class st_select_lex: public st_select_lex_node
   TABLE_LIST *embedding;          /* table embedding to the above list   */
   table_value_constr *tvc;
 
-  /* The interface employed to execute the select query by a foreign engine */
-  select_handler *select_h;
   /* The object used to organize execution of the query by a foreign engine */
   select_handler *pushdown_select;
   List<TABLE_LIST> *join_list;    /* list for the currently parsed join  */
@@ -1616,8 +1620,6 @@ class st_select_lex: public st_select_lex_node
                                        uchar *arg);
   Item *pushdown_from_having_into_where(THD *thd, Item *having);
 
-  select_handler *find_select_handler(THD *thd);
-
   bool is_set_op()
   {
     return linkage == UNION_TYPE || 

sql/sql_select.cc

@@ -71,6 +71,7 @@
 #include "derived_handler.h"
 #include "create_tmp_table.h"
 #include "optimizer_defaults.h"
+#include "derived_handler.h"
 
 /*
   A key part number that means we're using a fulltext scan.
@@ -1887,6 +1888,10 @@ int JOIN::optimize()
   join_optimization_state init_state= optimization_state;
   if (select_lex->pushdown_select)
   {
+    if (optimization_state == JOIN::OPTIMIZATION_DONE)
+      return 0;
+    DBUG_ASSERT(optimization_state == JOIN::NOT_OPTIMIZED);
+
     // Do same as JOIN::optimize_inner does:
     fields= &select_lex->item_list;
 
@@ -4980,9 +4985,20 @@ void JOIN::cleanup_item_list(List<Item> &items) const
 
 /**
   @brief
-    Look for provision of the select_handler interface by a foreign engine
-
-  @param thd   The thread handler
+    Look for provision of the select_handler interface by a foreign engine.
+    Must not be called directly, use find_single_select_handler() or
+    find_partial_select_handler() instead.
+
+  @param
+    thd             The thread handler
+    select_lex      SELECT_LEX object, must be passed in the cases of:
+                    - single select pushdown
+                    - partial pushdown (part of a UNION/EXCEPT/INTERSECT)
+                    Must be NULL in case of entire unit pushdown
+    select_lex_unit SELECT_LEX_UNIT object, must be passed in the cases of:
+                    - entire unit pushdown
+                    - partial pushdown (part of a UNION/EXCEPT/INTERSECT)
+                    Must be NULL in case of single select pushdown
 
   @details
     The function checks that this is an upper level select and if so looks
@@ -4990,18 +5006,21 @@ void JOIN::cleanup_item_list(List<Item> &items) const
     create_select call-back function. If the call of this function returns
     a select_handler interface object then the server will push the select
     query into this engine.
-    This is a responsibility of the create_select call-back function to
-    check whether the engine can execute the query.
+    This function does not check if the select has tables from
+    different engines. Such a check must be done inside each engine's
+    create_select function.
+    Also the engine's create_select function must perform other checks
+    to make sure the engine can execute the query.
 
   @retval the found select_handler if the search is successful
           0  otherwise
 */
 
-select_handler *find_select_handler(THD *thd,
-                                    SELECT_LEX* select_lex)
+static
+select_handler *find_select_handler_inner(THD *thd,
+                                    SELECT_LEX *select_lex,
+                                    SELECT_LEX_UNIT *select_lex_unit)
 {
-  if (select_lex->next_select())
-    return 0;
   if (select_lex->master_unit()->outer_select())
     return 0;
 
@@ -5028,13 +5047,47 @@ select_handler *find_select_handler(THD *thd,
     handlerton *ht= tbl->table->file->partition_ht();
     if (!ht->create_select)
       continue;
-    select_handler *sh= ht->create_select(thd, select_lex);
+    select_handler *sh= ht->create_select(thd, select_lex, select_lex_unit);
+    if (sh)
       return sh;
   }
   return 0;
 }
 
 
+/**
+  Wrapper for find_select_handler_inner() for the case of single select
+  pushdown. See more comments at the description of
+  find_select_handler_inner()
+
+*/
+select_handler *find_single_select_handler(THD *thd, SELECT_LEX *select_lex)
+{
+  return find_select_handler_inner(thd, select_lex, nullptr);
+}
+
+
+/**
+  Wrapper for find_select_handler_inner() for the case of partial select
+  pushdown. Partial pushdown means that a unit (i.e. multiple selects combined
+  with UNION/EXCEPT/INTERSECT operators) cannot be pushed down to
+  the storage engine as a whole but some particular selects of this unit can.
+  For example,
+    SELECT a FROM federated.t1  -- can be pushed down to Federated
+    UNION
+    SELECT b FROM local.t2      -- cannot be pushed down, executed locally
+
+  See more comments at the description of find_select_handler_inner()
+
+*/
+select_handler *
+find_partial_select_handler(THD *thd, SELECT_LEX *select_lex,
+                            SELECT_LEX_UNIT *select_lex_unit)
+{
+  return find_select_handler_inner(thd, select_lex, select_lex_unit);
+}
+
+
 /**
   An entry point to single-unit select (a select without UNION).
 
@@ -5144,7 +5197,7 @@ mysql_select(THD *thd, TABLE_LIST *tables, List<Item> &fields, COND *conds,
 
   thd->get_stmt_da()->reset_current_row_for_warning(1);
   /* Look for a table owned by an engine with the select_handler interface */
-  select_lex->pushdown_select= find_select_handler(thd, select_lex);
+  select_lex->pushdown_select= find_single_select_handler(thd, select_lex);
 
   if ((err= join->optimize()))
   {
@@ -15723,6 +15776,11 @@ void JOIN_TAB::cleanup()
       }
       DBUG_VOID_RETURN;
     }
+    /*if (table->pos_in_table_list && table->pos_in_table_list->derived)
+    {
+      delete table->pos_in_table_list->derived->derived->dt_handler;
+    }*/
+
     /*
       We need to reset this for next select
       (Tested in part_of_refkey)
@@ -30163,7 +30221,6 @@ bool mysql_explain_union(THD *thd, SELECT_LEX_UNIT *unit, select_result *result)
   DBUG_ENTER("mysql_explain_union");
   bool res= 0;
   SELECT_LEX *first= unit->first_select();
-  bool is_pushed_union= unit->derived && unit->derived->pushdown_derived;
 
   for (SELECT_LEX *sl= first; sl; sl= sl->next_select())
   {
@@ -30185,6 +30242,17 @@ bool mysql_explain_union(THD *thd, SELECT_LEX_UNIT *unit, select_result *result)
     if (!(res= unit->prepare(unit->derived, result,
                              SELECT_NO_UNLOCK | SELECT_DESCRIBE)))
     {
+      bool is_pushed_union=
+          (unit->derived && unit->derived->pushdown_derived) ||
+          unit->pushdown_unit;
+      if (unit->pushdown_unit)
+      {
+        create_explain_query_if_not_exists(thd->lex, thd->mem_root);
+        if (!unit->executed)
+          unit->save_union_explain(thd->lex->explain);
+        List<Item> items;
+        result->prepare(items, unit);
+      }
       if (!is_pushed_union)
         res= unit->exec();
     }
@@ -30744,7 +30812,7 @@ void st_select_lex::print(THD *thd, String *str, enum_query_type query_type)
 
   bool top_level= is_query_topmost(thd);
   enum explainable_cmd_type sel_type= SELECT_CMD;
-  if (top_level)
+  if (top_level && !(query_type & QT_SELECT_ONLY))
     sel_type= get_explainable_cmd_type(thd);
 
   if (sel_type == INSERT_CMD || sel_type == REPLACE_CMD)

sql/sql_union.cc

@@ -32,6 +32,9 @@
 #include "sql_cte.h"
 #include "item_windowfunc.h"
 
+select_handler *find_partial_select_handler(THD *thd, SELECT_LEX *select_lex,
+                                            SELECT_LEX_UNIT *lex_unit);
+
 bool mysql_union(THD *thd, LEX *lex, select_result *result,
                  SELECT_LEX_UNIT *unit, ulonglong setup_tables_done_option)
 {
@@ -1282,6 +1285,94 @@ bool init_item_int(THD* thd, Item_int* &item)
   return true;
 }
 
+/**
+  @brief
+    Recursive subroutine to be called from find_unit_handler() (see below).
+    Must not be called directly, only from find_unit_handler().
+*/
+static select_handler *find_unit_handler_for_lex(THD *thd,
+                                                 SELECT_LEX *sel_lex,
+                                                 SELECT_LEX_UNIT* unit)
+{
+  if (!(sel_lex->join))
+    return nullptr;
+  for (TABLE_LIST *tbl= sel_lex->join->tables_list; tbl; tbl= tbl->next_local)
+  {
+    if (!tbl->table)
+      continue;
+    if (tbl->derived)
+    {
+      /*
+        Skip derived table for now as they will be checked
+        in the subsequent loop
+      */
+      continue;
+    }
+    handlerton *ht= tbl->table->file->partition_ht();
+    if (!ht->create_unit)
+      continue;
+    select_handler *sh= ht->create_unit(thd, unit);
+    if (sh)
+      return sh;
+  }
+
+  for (SELECT_LEX_UNIT *un= sel_lex->first_inner_unit(); un;
+       un= un->next_unit())
+  {
+    for (SELECT_LEX *sl= un->first_select(); sl; sl= sl->next_select())
+    {
+      select_handler *uh= find_unit_handler_for_lex(thd, sl, unit);
+      if (uh)
+        return uh;
+    }
+  }
+  return nullptr;
+}
+
+
+/**
+  @brief
+    Look for provision of the select_handler interface by a foreign engine.
+    This interface must support processing UNITs (multiple SELECTs combined
+    with UNION/EXCEPT/INTERSECT operators)
+
+  @param
+    thd   The thread handler
+    unit  UNIT (one or more SELECTs combined with UNION/EXCEPT/INTERSECT
+
+  @details
+    The function checks that this is an upper level UNIT and if so looks
+    through its tables searching for one whose handlerton owns a
+    create_unit call-back function. If the call of this function returns
+    a select_handler interface object then the server will push the
+    query into this engine.
+    This is a responsibility of the create_unit call-back function to
+    check whether the engine can execute the query.
+
+    The function recursively scans subqueries (see find_unit_handler_for_lex())
+    to get down to real tables and process queries like this:
+      (SELECT a FROM t1 UNION SELECT b FROM t2) UNION
+        (SELECT c FROM t3 UNION select d FROM t4)
+
+  @retval the found select_handler if the search is successful
+          nullptr  otherwise
+*/
+
+static select_handler *find_unit_handler(THD *thd,
+                                         SELECT_LEX_UNIT *unit)
+{
+  if (unit->outer_select())
+    return nullptr;
+
+  for (SELECT_LEX *sl= unit->first_select(); sl; sl= sl->next_select())
+  {
+    select_handler *uh= find_unit_handler_for_lex(thd, sl, unit);
+    if (uh)
+      return uh;
+  }
+  return nullptr;
+}
+
 
 bool st_select_lex_unit::prepare(TABLE_LIST *derived_arg,
                                  select_result *sel_result,
@@ -1874,6 +1965,13 @@ bool st_select_lex_unit::prepare(TABLE_LIST *derived_arg,
     }
   }
 
+  pushdown_unit= find_unit_handler(thd, this);
+  if (pushdown_unit)
+  {
+    if (unlikely(pushdown_unit->prepare()))
+      DBUG_RETURN(TRUE);
+  }
+
   thd->lex->current_select= lex_select_save;
 
   DBUG_RETURN(saved_error || thd->is_fatal_error);
@@ -2133,6 +2231,15 @@ bool st_select_lex_unit::optimize()
           (lim.is_unlimited() || sl->braces) ?
           sl->options & ~OPTION_FOUND_ROWS : sl->options | found_rows_for_union;
 
+        if (!this->pushdown_unit)
+        {
+          /*
+            If the UNIT hasn't been pushed down to the engine as a whole,
+            try to push down partial SELECTs of this UNIT separately
+          */
+          sl->pushdown_select= find_partial_select_handler(thd, sl, this);
+        }
+
 	saved_error= sl->join->optimize();
       }
 
@@ -2151,6 +2258,23 @@ bool st_select_lex_unit::optimize()
 
 
 bool st_select_lex_unit::exec()
+{
+  DBUG_ENTER("st_select_lex_unit::exec");
+  if (executed && !uncacheable && !describe)
+    DBUG_RETURN(FALSE);
+
+  if (pushdown_unit)
+  {
+    create_explain_query_if_not_exists(thd->lex, thd->mem_root);
+    if (!executed)
+      save_union_explain(thd->lex->explain);
+    DBUG_RETURN(pushdown_unit->execute());
+  }
+  DBUG_RETURN(exec_inner());
+}
+
+
+bool st_select_lex_unit::exec_inner()
 {
   SELECT_LEX *lex_select_save= thd->lex->current_select;
   SELECT_LEX *select_cursor=first_select();
@@ -2158,10 +2282,7 @@ bool st_select_lex_unit::exec()
   ha_rows examined_rows= 0;
   bool first_execution= !executed;
   bool was_executed= executed;
-  DBUG_ENTER("st_select_lex_unit::exec");
 
-  if (executed && !uncacheable && !describe)
-    DBUG_RETURN(FALSE);
   executed= 1;
   if (!(uncacheable & ~UNCACHEABLE_EXPLAIN) && item &&
       !item->with_recursive_reference)
@@ -2175,7 +2296,7 @@ bool st_select_lex_unit::exec()
     save_union_explain(thd->lex->explain);
 
   if (unlikely(saved_error))
-    DBUG_RETURN(saved_error);
+    return saved_error;
 
   if (union_result)
   {
@@ -2192,6 +2313,7 @@ bool st_select_lex_unit::exec()
   {
     if (!fake_select_lex && !(with_element && with_element->is_recursive))
       union_result->cleanup();
+
     for (SELECT_LEX *sl= select_cursor; sl; sl= sl->next_select())
     {
       ha_rows records_at_start= 0;
@@ -2252,7 +2374,7 @@ bool st_select_lex_unit::exec()
           // This is UNION DISTINCT, so there should be a fake_select_lex
           DBUG_ASSERT(fake_select_lex != NULL);
           if (unlikely(table->file->ha_disable_indexes(HA_KEY_SWITCH_ALL)))
-	    DBUG_RETURN(TRUE);
+            return true;
 	  table->no_keyread=1;
 	}
 	if (likely(!saved_error))
@@ -2262,14 +2384,14 @@ bool st_select_lex_unit::exec()
 	  if (union_result->flush())
 	  {
 	    thd->lex->current_select= lex_select_save;
-	    DBUG_RETURN(1);
+	    return true;
 	  }
 	}
       }
       if (unlikely(saved_error))
       {
 	thd->lex->current_select= lex_select_save;
-	DBUG_RETURN(saved_error);
+	return saved_error;
       }
       if (fake_select_lex != NULL)
       {
@@ -2278,7 +2400,7 @@ bool st_select_lex_unit::exec()
         if (unlikely(error))
         {
           table->file->print_error(error, MYF(0));
-          DBUG_RETURN(1);
+          return true;
         }
       }
       if (found_rows_for_union && !sl->braces &&
@@ -2421,7 +2543,7 @@ bool st_select_lex_unit::exec()
   thd->lex->current_select= lex_select_save;
 err:
   thd->lex->set_limit_rows_examined();
-  DBUG_RETURN(saved_error);
+  return saved_error;
 }
 
 
@@ -2648,6 +2770,9 @@ bool st_select_lex_unit::cleanup()
     }
   }
 
+  delete pushdown_unit;
+  pushdown_unit= nullptr;
+
   DBUG_RETURN(error);
 }
 
@@ -2811,6 +2936,8 @@ bool st_select_lex::cleanup()
   inner_refs_list.empty();
   exclude_from_table_unique_test= FALSE;
   hidden_bit_fields= 0;
+  delete pushdown_select;
+  pushdown_select= nullptr;
   DBUG_RETURN(error);
 }
 

sql/table.h

@@ -2540,8 +2540,6 @@ struct TABLE_LIST
   bool block_handle_derived;
   /* The interface employed to materialize the table by a foreign engine */
   derived_handler *dt_handler;
-  /* The text of the query specifying the derived table */
-  LEX_CSTRING derived_spec;
   /*
     The object used to organize execution of the query that specifies
     the derived table by a foreign engine

storage/federatedx/federatedx_pushdown.h

@@ -14,29 +14,43 @@
    along with this program; if not, write to the Free Software
    Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA */
 
+#include "sql_string.h"
 #include "derived_handler.h"
 #include "select_handler.h"
 
+class federatedx_handler_base
+{
+protected:
+  FEDERATEDX_SHARE *share;
+  federatedx_txn *txn;
+  federatedx_io **iop;
+  FEDERATEDX_IO_RESULT *stored_result;
+  StringBuffer<512> query;
+  
+  TABLE *query_table;
+
+  int next_row_(TABLE *tmp_tbl);
+  int init_scan_();
+  int end_scan_();
+public:
+  federatedx_handler_base(THD *thd_arg, TABLE *tbl_arg);
+};
+
 /*
   Implementation class of the derived_handler interface for FEDERATEDX:
   class declaration
 */
 
-class ha_federatedx_derived_handler: public derived_handler
+class ha_federatedx_derived_handler: public derived_handler, public federatedx_handler_base
 {
 private:
-  FEDERATEDX_SHARE *share;
-  federatedx_txn *txn;
-  federatedx_io **iop;
-  FEDERATEDX_IO_RESULT *stored_result;
 
 public:
-  ha_federatedx_derived_handler(THD* thd_arg, TABLE_LIST *tbl);
+  ha_federatedx_derived_handler(THD* thd_arg, TABLE_LIST *tbl, TABLE *tbl_arg);
   ~ha_federatedx_derived_handler();
-  int init_scan();
-  int next_row();
-  int end_scan();
-  void print_error(int, unsigned long);
+  int init_scan() { return federatedx_handler_base::init_scan_(); }
+  int next_row() { return federatedx_handler_base::next_row_(table); }
+  int end_scan() { return federatedx_handler_base::end_scan_(); }
 };
 
 
@@ -45,19 +59,17 @@ class ha_federatedx_derived_handler: public derived_handler
   class declaration
 */
 
-class ha_federatedx_select_handler: public select_handler
+class ha_federatedx_select_handler: public select_handler, public federatedx_handler_base
 {
-private:
-  FEDERATEDX_SHARE *share;
-  federatedx_txn *txn;
-  federatedx_io **iop;
-  FEDERATEDX_IO_RESULT *stored_result;
-
 public:
-  ha_federatedx_select_handler(THD* thd_arg, SELECT_LEX *sel);
+  ha_federatedx_select_handler(THD *thd_arg, SELECT_LEX *sel_lex,
+                               TABLE *tbl);
+  ha_federatedx_select_handler(THD *thd_arg, SELECT_LEX_UNIT *sel_unit,
+                               TABLE *tbl);
+  ha_federatedx_select_handler(THD *thd_arg, SELECT_LEX *sel_lex,
+                               SELECT_LEX_UNIT *sel_unit, TABLE *tbl);
   ~ha_federatedx_select_handler();
-  int init_scan();
-  int next_row();
+  int init_scan() { return federatedx_handler_base::init_scan_(); }
+  int next_row() { return federatedx_handler_base::next_row_(table); }
   int end_scan();
-  void print_error(int, unsigned long);
 };

storage/federatedx/ha_federatedx.cc

@@ -408,8 +408,12 @@ handlerton* federatedx_hton;
 
 static derived_handler*
 create_federatedx_derived_handler(THD* thd, TABLE_LIST *derived);
+
 static select_handler*
-create_federatedx_select_handler(THD* thd, SELECT_LEX *sel);
+create_federatedx_select_handler(THD *thd, SELECT_LEX *sel_lex,
+                                 SELECT_LEX_UNIT *sel_unit);
+static select_handler *
+create_federatedx_unit_handler(THD *thd, SELECT_LEX_UNIT *sel_unit);
 
 /*
   Federated doesn't need costs.disk_read_ratio as everything is one a remote
@@ -458,6 +462,7 @@ int federatedx_db_init(void *p)
   federatedx_hton->create_derived= create_federatedx_derived_handler;
   federatedx_hton->create_select= create_federatedx_select_handler;
   federatedx_hton->update_optimizer_costs= federatedx_update_optimizer_costs;
+  federatedx_hton->create_unit= create_federatedx_unit_handler;
 
   if (mysql_mutex_init(fe_key_mutex_federatedx,
                        &federatedx_mutex, MY_MUTEX_INIT_FAST))

storage/federatedx/ha_federatedx.h

@@ -467,6 +467,7 @@ class ha_federatedx final : public handler
   const FEDERATEDX_SHARE *get_federatedx_share() const { return share; }
   friend class ha_federatedx_derived_handler;
   friend class ha_federatedx_select_handler;
+  friend class federatedx_handler_base;
 };
 
 extern const char ident_quote_char;              // Character for quoting