MDEV-25080 Implement pushdown of SELECT_UNITsbb-10.10-MDEV-25080

13 files changed, 414 insertions, 112 deletions

diff --git a/sql/handler.h b/sql/handler.h
index 60c6195e68e..8630b8883c3 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -54,6 +54,7 @@ class Field_string;
 class Field_varstring;
 class Field_blob;
 class Column_definition;
+class select_result;
 
 // the following is for checking tables
 
@@ -1203,6 +1204,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;
 
 /*
@@ -1544,10 +1546,17 @@ 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);
+
+  /*
+    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.
diff --git a/sql/select_handler.cc b/sql/select_handler.cc
index 795ed8eb641..99bd2c0cf7f 100644
--- a/sql/select_handler.cc
+++ b/sql/select_handler.cc
@@ -36,9 +36,18 @@
 */
 
 
-select_handler::select_handler(THD *thd_arg, handlerton *ht_arg)
- : thd(thd_arg), ht(ht_arg), table(NULL),
-   is_analyze(thd_arg->lex->analyze_stmt)
+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)
 {}
 
 
@@ -49,13 +58,18 @@ select_handler::~select_handler()
 }
 
 
-TABLE *select_handler::create_tmp_table(THD *thd, SELECT_LEX *select)
+TABLE *select_handler::create_tmp_table(THD *thd)
 {
   DBUG_ENTER("select_handler::create_tmp_table");
   List<Item> types;
   TMP_TABLE_PARAM tmp_table_param;
-  if (select->master_unit()->join_union_item_types(thd, types, 1))
+  
+  if (select_lex &&
+      select_lex->master_unit()->join_union_item_types(thd, types, 1))
+    DBUG_RETURN(NULL);
+  else if (lex_unit && lex_unit->join_union_item_types(thd, types, 1))
     DBUG_RETURN(NULL);
+
   tmp_table_param.init();
   tmp_table_param.field_count= types.elements;
 
@@ -74,7 +88,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));
 }
@@ -91,22 +105,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);
 }
 
@@ -114,10 +125,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());
 }
 
 
diff --git a/sql/select_handler.h b/sql/select_handler.h
index 5cc63231641..59f36e2e4c2 100644
--- a/sql/select_handler.h
+++ b/sql/select_handler.h
@@ -30,10 +30,25 @@
 class select_handler
 {
  public:
-  THD *thd;
-  handlerton *ht;
+  select_handler(THD *thd_arg, handlerton *ht_arg, SELECT_LEX *sel_lex);
+  select_handler(THD *thd_arg, handlerton *ht_arg, SELECT_LEX_UNIT *sel_unit);
 
-  SELECT_LEX *select;  // Select to be excuted
+  virtual ~select_handler();
+
+  int execute();
+
+  virtual bool prepare();
+
+  /*
+    Select_handler processes either a single SELECT or a UNIT
+    (multiple SELECTs combined with UNION/EXCEPT/INTERSECT).
+    Only one of the following two class members is initialized while another
+    one is strictly NULL. In case of a single SELECT select_lex is initialized
+    and lex_unit == NULL, in case of UNIT select_lex == NULL
+    and lex_unit is 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]
@@ -41,24 +56,9 @@ class select_handler
     The table is actually never filled. Only its record buffer is used.
   */
   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);
-
-  virtual ~select_handler();
-
-  int execute();
-
-  virtual bool prepare();
-
-  static TABLE *create_tmp_table(THD *thd, SELECT_LEX *sel);
 
 protected:
+
   /*
     Functions to scan the select result set.
     All these returns 0 if ok, error code in case of error.
@@ -80,7 +80,19 @@ protected:
   /* 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 */
diff --git a/sql/sql_derived.cc b/sql/sql_derived.cc
index 579ea34b8e4..ece0bbe5363 100644
--- a/sql/sql_derived.cc
+++ b/sql/sql_derived.cc
@@ -904,6 +904,7 @@ bool mysql_derived_prepare(THD *thd, LEX *lex, TABLE_LIST *derived)
   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,
diff --git a/sql/sql_explain.cc b/sql/sql_explain.cc
index 7c3918bfd20..90e9635fe5a 100644
--- a/sql/sql_explain.cc
+++ b/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";
 
@@ -538,7 +543,22 @@ uint Explain_union::make_union_table_name(char *buf)
 }
 
 
-int Explain_union::print_explain(Explain_query *query, 
+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)
@@ -555,6 +575,12 @@ 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" */
@@ -566,12 +592,11 @@ int Explain_union::print_explain(Explain_query *query,
 
   /* `select_type` column */
   push_str(thd, &item_list, fake_select_type);
-
+  
   /* `table` column: something like "<union1,2>" */
   uint len= make_union_table_name(table_name_buffer);
-  item_list.push_back(new (mem_root)
-                      Item_string_sys(thd, table_name_buffer, len),
-                      mem_root);
+  item_list.push_back(
+      new (mem_root) Item_string_sys(thd, table_name_buffer, len), mem_root);
   
   /* `partitions` column */
   if (explain_flags & DESCRIBE_PARTITIONS)
@@ -628,7 +653,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;
   
@@ -640,10 +664,90 @@ int Explain_union::print_explain(Explain_query *query,
 }
 
 
-void Explain_union::print_explain_json(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];
   
@@ -701,6 +805,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("Pushed union");
+
+  writer->end_object(); // union_result
+  writer->end_object(); // query_block
+}
+
 
 /*
   Print EXPLAINs for all children nodes (i.e. for subqueries)
diff --git a/sql/sql_explain.h b/sql/sql_explain.h
index 919d51aef8b..f8eba0b53d4 100644
--- a/sql/sql_explain.h
+++ b/sql/sql_explain.h
@@ -334,6 +334,7 @@ public:
 /////////////////////////////////////////////////////////////////////////////
 
 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;
 
@@ -348,7 +349,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)
   {}
 
@@ -381,11 +382,17 @@ public:
                     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
@@ -404,6 +411,10 @@ public:
   }
 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 */
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index 9382625a747..b02a7a0270a 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -5495,12 +5495,18 @@ void st_select_lex::set_explain_type(bool on_the_fly)
       using_materialization= TRUE;
   }
 
+  if (!on_the_fly)
+    options|= SELECT_DESCRIBE;
+
+  if (pushdown_select)
+  {
+    type= pushed_select_text;
+    return;
+  }
+  
   if (master_unit()->thd->lex->first_select_lex() == this)
   {
-    if (pushdown_select)
-      type= pushed_select_text;
-    else
-      type= is_primary ? "PRIMARY" : "SIMPLE";
+    type= is_primary ? "PRIMARY" : "SIMPLE";
   }
   else
   {
@@ -5510,7 +5516,7 @@ void st_select_lex::set_explain_type(bool on_the_fly)
       if (linkage == DERIVED_TABLE_TYPE)
       {
         bool is_pushed_master_unit= master_unit()->derived &&
-	                            master_unit()->derived->pushdown_derived;
+                                    master_unit()->derived->pushdown_derived;
         if (is_pushed_master_unit)
           type= pushed_derived_text;
         else if (is_uncacheable & UNCACHEABLE_DEPENDENT)
@@ -5522,13 +5528,12 @@ void st_select_lex::set_explain_type(bool on_the_fly)
         type= "MATERIALIZED";
       else
       {
-         if (is_uncacheable & UNCACHEABLE_DEPENDENT)
-           type= "DEPENDENT SUBQUERY";
-         else
-         {
-           type= is_uncacheable? "UNCACHEABLE SUBQUERY" :
-                                 "SUBQUERY";
-         }
+        if (is_uncacheable & UNCACHEABLE_DEPENDENT)
+          type= "DEPENDENT SUBQUERY";
+        else
+        {
+          type= is_uncacheable ? "UNCACHEABLE SUBQUERY" : "SUBQUERY";
+        }
       }
     }
     else
@@ -5549,9 +5554,13 @@ void st_select_lex::set_explain_type(bool on_the_fly)
           type= "MATERIALIZED UNION";
         else
         {
-          type= is_uncacheable ? "UNCACHEABLE UNION": "UNION";
+          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
@@ -5561,15 +5570,15 @@ void st_select_lex::set_explain_type(bool on_the_fly)
           {
             bool uses_cte= false;
             for (JOIN_TAB *tab= first_linear_tab(join, WITHOUT_BUSH_ROOTS,
-                                                       WITH_CONST_TABLES);
-                 tab;
-                 tab= next_linear_tab(join, tab, WITHOUT_BUSH_ROOTS))
+                                                  WITH_CONST_TABLES);
+                  tab; tab= next_linear_tab(join, tab, WITHOUT_BUSH_ROOTS))
             {
               /*
-                pos_in_table_list=NULL for e.g. post-join aggregation JOIN_TABs.
+                pos_in_table_list=NULL for e.g. post-join aggregation
+                JOIN_TABs.
               */
               if (!(tab->table && tab->table->pos_in_table_list))
-	        continue;
+                continue;
               TABLE_LIST *tbl= tab->table->pos_in_table_list;
               if (tbl->with && tbl->with->is_recursive &&
                   tbl->is_with_table_recursive_reference())
@@ -5586,9 +5595,6 @@ void st_select_lex::set_explain_type(bool on_the_fly)
       }
     }
   }
-
-  if (!on_the_fly)
-    options|= SELECT_DESCRIBE;
 }
 
 
@@ -6011,7 +6017,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();
 
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index e58db8fa502..15c8f55ae5d 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -36,6 +36,7 @@
 #include "sql_limit.h"                // Select_limit_counters
 #include "json_table.h"               // Json_table_column
 #include "sql_schema.h"
+#include "select_handler.h"
 
 /* Used for flags of nesting constructs */
 #define SELECT_NESTING_MAP_SIZE 64
@@ -870,7 +871,7 @@ public:
   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)
   {
   }
 
@@ -933,6 +934,10 @@ public:
   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:
@@ -1042,6 +1047,7 @@ public:
   friend class st_select_lex;
 
 private:
+  bool exec_inner();
   bool is_derived_eliminated() const;
 };
 
@@ -1137,8 +1143,6 @@ public:
   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  */
@@ -1595,8 +1599,6 @@ public:
                                        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 || 
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index b76ec6d14f3..ab057d79dc3 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -1842,6 +1842,8 @@ int JOIN::optimize()
   join_optimization_state init_state= optimization_state;
   if (select_lex->pushdown_select)
   {
+    if (optimization_state == JOIN::OPTIMIZATION_DONE)
+      return 0;
     // Do same as JOIN::optimize_inner does:
     fields= &select_lex->item_list;
 
@@ -4884,7 +4886,6 @@ void JOIN::cleanup_item_list(List<Item> &items) const
   DBUG_VOID_RETURN;
 }
 
-
 /**
   @brief
     Look for provision of the select_handler interface by a foreign engine
@@ -4897,18 +4898,18 @@ 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)
+select_handler *find_select_handler(THD *thd, SELECT_LEX *select_lex)
 {
-  if (select_lex->next_select())
-    return 0;
   if (select_lex->master_unit()->outer_select())
     return 0;
 
@@ -4920,15 +4921,14 @@ select_handler *find_select_handler(THD *thd,
   {
     tbl= select_lex->join->tables_list;
   }
-  else if (thd->lex->query_tables &&
-           thd->lex->query_tables->next_global)
+  else if (thd->lex->query_tables && thd->lex->query_tables->next_global)
   {
     tbl= thd->lex->query_tables->next_global;
   }
   else
     return 0;
 
-  for (;tbl; tbl= tbl->next_global)
+  for (; tbl; tbl= tbl->next_global)
   {
     if (!tbl->table)
       continue;
@@ -4936,12 +4936,14 @@ select_handler *find_select_handler(THD *thd,
     if (!ht->create_select)
       continue;
     select_handler *sh= ht->create_select(thd, select_lex);
-    return sh;
+    if (sh)
+      return sh;
   }
   return 0;
 }
 
 
+
 /**
   An entry point to single-unit select (a select without UNION).
 
@@ -28456,7 +28458,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())
   {
@@ -28478,6 +28479,18 @@ 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();
     }
diff --git a/sql/sql_union.cc b/sql/sql_union.cc
index e8bdbcba2f2..bc99dc497e9 100644
--- a/sql/sql_union.cc
+++ b/sql/sql_union.cc
@@ -1294,6 +1294,52 @@ bool init_item_int(THD* thd, Item_int* &item)
   return true;
 }
 
+/**
+  @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
+
+  @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.
+
+  @retval the found select_handler if the search is successful
+          0  otherwise
+*/
+
+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())
+  {
+    if (!(sl->join))
+      continue;
+    for (TABLE_LIST *tbl= sl->join->tables_list; tbl; tbl= tbl->next_local)
+    {
+      if (!tbl->table)
+        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;
+    }
+  }
+  return nullptr;
+}
+
 
 bool st_select_lex_unit::prepare(TABLE_LIST *derived_arg,
                                  select_result *sel_result,
@@ -1882,6 +1928,13 @@ cont:
     }
   }
 
+  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);
@@ -2053,6 +2106,7 @@ void st_select_lex_unit::optimize_bag_operation(bool is_outer_distinct)
   bag_set_op_optimized= true;
 }
 
+select_handler *find_select_handler(THD *thd, SELECT_LEX *select_lex);
 
 /**
   Run optimization phase.
@@ -2141,6 +2195,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_select_handler(thd, sl);
+        }
+
 	saved_error= sl->join->optimize();
       }
 
@@ -2158,18 +2221,33 @@ bool st_select_lex_unit::optimize()
 }
 
 
-bool st_select_lex_unit::exec()
+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();
   ulonglong add_rows=0;
   ha_rows examined_rows= 0;
   bool first_execution= !executed;
-  DBUG_ENTER("st_select_lex_unit::exec");
   bool was_executed= executed;
 
-  if (executed && !uncacheable && !describe)
-    DBUG_RETURN(FALSE);
   executed= 1;
   if (!(uncacheable & ~UNCACHEABLE_EXPLAIN) && item &&
       !item->with_recursive_reference)
@@ -2183,7 +2261,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)
   {
@@ -2200,6 +2278,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;
@@ -2259,8 +2338,8 @@ 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);
+          if (unlikely(table->file->ha_disable_indexes(HA_KEY_SWITCH_ALL)))
+            return true;
 	  table->no_keyread=1;
 	}
 	if (!sl->tvc)
@@ -2272,14 +2351,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)
       {
@@ -2288,7 +2367,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 &&
@@ -2430,7 +2509,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;
 }
 
 
@@ -2659,6 +2738,9 @@ bool st_select_lex_unit::cleanup()
     }
   }
 
+  delete pushdown_unit;
+  pushdown_unit= nullptr;
+  
   DBUG_RETURN(error);
 }
 
diff --git a/storage/federatedx/federatedx_pushdown.cc b/storage/federatedx/federatedx_pushdown.cc
index 664f0570238..0fa7329d800 100644
--- a/storage/federatedx/federatedx_pushdown.cc
+++ b/storage/federatedx/federatedx_pushdown.cc
@@ -163,13 +163,13 @@ void ha_federatedx_derived_handler::print_error(int, unsigned long)
 }
 
 
-static select_handler*
-create_federatedx_select_handler(THD* thd, SELECT_LEX *sel)
+template<typename T>
+static select_handler *create_federatedx_handler(THD *thd, T *sel_lex)
 {
   if (!use_pushdown)
     return 0;
 
-  ha_federatedx_select_handler* handler = NULL;
+  ha_federatedx_select_handler *handler= NULL;
   handlerton *ht= 0;
 
   for (TABLE_LIST *tbl= thd->lex->query_tables; tbl; tbl= tbl->next_global)
@@ -189,27 +189,48 @@ create_federatedx_select_handler(THD* thd, SELECT_LEX *sel)
     "INTO OUTFILE". It is also unlikely to work if the select has some
     other kind of side effect.
   */
-  if (sel->uncacheable & UNCACHEABLE_SIDEEFFECT)
+  if (sel_lex->uncacheable & UNCACHEABLE_SIDEEFFECT)
     return NULL;
 
-  handler= new ha_federatedx_select_handler(thd, sel);
+  handler= new ha_federatedx_select_handler(thd, sel_lex);
 
   return handler;
 }
 
+static select_handler *create_federatedx_select_handler(
+  THD *thd, SELECT_LEX *sel_lex)
+{
+  return create_federatedx_handler(thd, sel_lex);
+}
+
+static select_handler *create_federatedx_unit_handler(
+  THD* thd, SELECT_LEX_UNIT *sel_unit)
+{
+  return create_federatedx_handler(thd, sel_unit);
+}
+
 /*
   Implementation class of the select_handler interface for FEDERATEDX:
   class implementation
 */
 
-ha_federatedx_select_handler::ha_federatedx_select_handler(THD *thd,
-                                                           SELECT_LEX *sel)
-  : select_handler(thd, federatedx_hton),
+
+ha_federatedx_select_handler::ha_federatedx_select_handler(
+    THD *thd, SELECT_LEX *select_lex)
+    : select_handler(thd, federatedx_hton, select_lex), 
+      share(NULL), txn(NULL), iop(NULL), stored_result(NULL)
+{
+}
+
+ha_federatedx_select_handler::ha_federatedx_select_handler(
+    THD *thd,
+    SELECT_LEX_UNIT *lex_unit)
+  : select_handler(thd, federatedx_hton, lex_unit),
     share(NULL), txn(NULL), iop(NULL), stored_result(NULL)
 {
-  select= sel;
 }
 
+
 ha_federatedx_select_handler::~ha_federatedx_select_handler() {}
 
 int ha_federatedx_select_handler::init_scan()
diff --git a/storage/federatedx/federatedx_pushdown.h b/storage/federatedx/federatedx_pushdown.h
index 673abcfc68d..c0ed0da168d 100644
--- a/storage/federatedx/federatedx_pushdown.h
+++ b/storage/federatedx/federatedx_pushdown.h
@@ -54,7 +54,8 @@ private:
   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);
+  ha_federatedx_select_handler(THD *thd_arg, SELECT_LEX_UNIT *sel_unit);
   ~ha_federatedx_select_handler();
   int init_scan();
   int next_row();
diff --git a/storage/federatedx/ha_federatedx.cc b/storage/federatedx/ha_federatedx.cc
index d4ac808add9..db8ad391b4e 100644
--- a/storage/federatedx/ha_federatedx.cc
+++ b/storage/federatedx/ha_federatedx.cc
@@ -407,8 +407,11 @@ 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);
+static select_handler *
+create_federatedx_unit_handler(THD *thd, SELECT_LEX_UNIT *sel_unit);
 
 /*
   Initialize the federatedx handler.
@@ -442,6 +445,7 @@ int federatedx_db_init(void *p)
   federatedx_hton->flags= HTON_ALTER_NOT_SUPPORTED;
   federatedx_hton->create_derived= create_federatedx_derived_handler;
   federatedx_hton->create_select= create_federatedx_select_handler;
+  federatedx_hton->create_unit= create_federatedx_unit_handler;
 
   if (mysql_mutex_init(fe_key_mutex_federatedx,
                        &federatedx_mutex, MY_MUTEX_INIT_FAST))