MDEV-334: Backport of UNION ALL optimization from mysql-5.7.

Although the original code of mysql-5.7 was adjusted
to the current MariaDB code the main ideas of the optimization
were preserved.

13 files changed, 600 insertions, 226 deletions

diff --git a/sql/item_subselect.cc b/sql/item_subselect.cc
index 38bb3121ed8..affea8d28f8 100644
--- a/sql/item_subselect.cc
+++ b/sql/item_subselect.cc
@@ -1430,7 +1430,7 @@ void Item_exists_subselect::fix_length_and_dec()
     We need only 1 row to determine existence (i.e. any EXISTS that is not
     an IN always requires LIMIT 1)
   */
-  thd->change_item_tree(&unit->global_parameters->select_limit,
+  thd->change_item_tree(&unit->global_parameters()->select_limit,
                         new Item_int((int32) 1));
   DBUG_PRINT("info", ("Set limit to 1"));
   DBUG_VOID_RETURN;
@@ -2540,7 +2540,7 @@ bool Item_in_subselect::inject_in_to_exists_cond(JOIN *join_arg)
     select_lex->having->top_level_item();
     join_arg->having= select_lex->having;
   }
-  join_arg->thd->change_item_tree(&unit->global_parameters->select_limit,
+  join_arg->thd->change_item_tree(&unit->global_parameters()->select_limit,
                                   new Item_int((int32) 1));
   unit->select_limit_cnt= 1;
 
@@ -3593,7 +3593,7 @@ int subselect_single_select_engine::exec()
   {
     SELECT_LEX_UNIT *unit= select_lex->master_unit();
 
-    unit->set_limit(unit->global_parameters);
+    unit->set_limit(unit->global_parameters());
     if (join->optimize())
     {
       thd->where= save_where;
@@ -4310,7 +4310,7 @@ subselect_single_select_engine::change_result(Item_subselect *si,
     that would not require a lot of extra code that would be harder to manage
     than the current code.
   */
-  DBUG_RETURN(select_lex->join->change_result(res));
+  DBUG_RETURN(select_lex->join->change_result(res, NULL));
 }
 
 
@@ -4820,7 +4820,7 @@ bool subselect_hash_sj_engine::init(List<Item> *tmp_columns, uint subquery_id)
     DBUG_RETURN(TRUE);
   /* Let our engine reuse this query plan for materialization. */
   materialize_join= materialize_engine->join;
-  materialize_join->change_result(result);
+  materialize_join->change_result(result, NULL);
 
   DBUG_RETURN(FALSE);
 }
diff --git a/sql/opt_subselect.cc b/sql/opt_subselect.cc
index 4902ae8b028..66b1ba16cbc 100644
--- a/sql/opt_subselect.cc
+++ b/sql/opt_subselect.cc
@@ -5224,7 +5224,7 @@ bool setup_jtbm_semi_joins(JOIN *join, List<TABLE_LIST> *join_list,
         if (!(new_sink= new select_value_catcher(subq_pred)))
           DBUG_RETURN(TRUE);
         if (new_sink->setup(&engine->select_lex->join->fields_list) ||
-            engine->select_lex->join->change_result(new_sink) ||
+            engine->select_lex->join->change_result(new_sink, NULL) ||
             engine->exec())
         {
           DBUG_RETURN(TRUE);
@@ -5576,8 +5576,8 @@ bool JOIN::choose_subquery_plan(table_map join_tables)
       Item_in_subselect::test_limit). However, once we allow this, here
       we should set the correct limit if given in the query.
     */
-    in_subs->unit->global_parameters->select_limit= NULL;
-    in_subs->unit->set_limit(unit->global_parameters);
+    in_subs->unit->global_parameters()->select_limit= NULL;
+    in_subs->unit->set_limit(unit->global_parameters());
     /*
       Set the limit of this JOIN object as well, because normally its being
       set in the beginning of JOIN::optimize, which was already done.
diff --git a/sql/sql_class.h b/sql/sql_class.h
index d7bbfc3799d..9fe524b5fe6 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -3914,6 +3914,23 @@ protected:
 public:
   select_result();
   virtual ~select_result() {};
+  /**
+    Change wrapped select_result.
+
+    Replace the wrapped result object with new_result and call
+    prepare() and prepare2() on new_result.
+
+    This base class implementation doesn't wrap other select_results.
+
+    @param new_result The new result object to wrap around
+
+    @retval false Success
+    @retval true  Error
+  */
+  virtual bool change_result(select_result *new_result)
+  {
+    return false;
+  }
   virtual int prepare(List<Item> &list, SELECT_LEX_UNIT *u)
   {
     unit= u;
@@ -4321,9 +4338,19 @@ public:
 
   select_union() :write_err(0), table(0), records(0) { tmp_table_param.init(); }
   int prepare(List<Item> &list, SELECT_LEX_UNIT *u);
+  /**
+    Do prepare() and prepare2() if they have been postponed until
+    column type information is computed (used by select_union_direct).
+
+    @param types Column types
+
+    @return false on success, true on failure
+  */
+  virtual bool postponed_prepare(List<Item> &types)
+  { return false; }
   int send_data(List<Item> &items);
   bool send_eof();
-  bool flush();
+  virtual bool flush();
   void cleanup();
   virtual bool create_result_table(THD *thd, List<Item> *column_types,
                                    bool is_distinct, ulonglong options,
@@ -4334,6 +4361,101 @@ public:
   TMP_TABLE_PARAM *get_tmp_table_param() { return &tmp_table_param; }
 };
 
+
+/**
+  UNION result that is passed directly to the receiving select_result
+  without filling a temporary table.
+
+  Function calls are forwarded to the wrapped select_result, but some
+  functions are expected to be called only once for each query, so
+  they are only executed for the first SELECT in the union (execept
+  for send_eof(), which is executed only for the last SELECT).
+
+  This select_result is used when a UNION is not DISTINCT and doesn't
+  have a global ORDER BY clause. @see st_select_lex_unit::prepare().
+*/
+
+class select_union_direct :public select_union
+{
+private:
+  /* Result object that receives all rows */
+  select_result *result;
+  /* The last SELECT_LEX of the union */
+  SELECT_LEX *last_select_lex;
+
+  /* Wrapped result has received metadata */
+  bool done_send_result_set_metadata;
+  /* Wrapped result has initialized tables */
+  bool done_initialize_tables;
+
+  /* Accumulated limit_found_rows */
+  ulonglong limit_found_rows;
+
+  /* Number of rows offset */
+  ha_rows offset;
+  /* Number of rows limit + offset, @see select_union_direct::send_data() */
+  ha_rows limit;
+
+public:
+  select_union_direct(select_result *result, SELECT_LEX *last_select_lex)
+    :result(result), last_select_lex(last_select_lex),
+    done_send_result_set_metadata(false), done_initialize_tables(false),
+    limit_found_rows(0)
+  {}
+  bool change_result(select_result *new_result);
+  uint field_count(List<Item> &fields) const
+  {
+    // Only called for top-level select_results, usually select_send
+    DBUG_ASSERT(false); /* purecov: inspected */
+    return 0; /* purecov: inspected */
+  }
+  bool postponed_prepare(List<Item> &types);
+  bool send_result_set_metadata(List<Item> &list, uint flags);
+  int send_data(List<Item> &items);
+  bool initialize_tables (JOIN *join= NULL);
+  bool send_eof();
+  bool flush() { return false; }
+  bool check_simple_select() const
+  {
+    /* Only called for top-level select_results, usually select_send */
+    DBUG_ASSERT(false); /* purecov: inspected */
+    return false; /* purecov: inspected */
+  }
+  void abort_result_set()
+  {
+    result->abort_result_set(); /* purecov: inspected */
+  }
+  void cleanup()
+  {
+    /*
+      Only called for top-level select_results, usually select_send,
+      and for the results of subquery engines
+      (select_<something>_subselect).
+    */
+    DBUG_ASSERT(false); /* purecov: inspected */
+  }
+  void set_thd(THD *thd_arg)
+  {
+    /*
+      Only called for top-level select_results, usually select_send,
+      and for the results of subquery engines
+      (select_<something>_subselect).
+    */
+    DBUG_ASSERT(false); /* purecov: inspected */
+  }
+  void reset_offset_limit_cnt()
+  {
+    // EXPLAIN should never output to a select_union_direct
+    DBUG_ASSERT(false); /* purecov: inspected */
+  }
+  void begin_dataset()
+  {
+    // Only called for sp_cursor::Select_fetch_into_spvars
+    DBUG_ASSERT(false); /* purecov: inspected */
+  }
+};
+
+
 /* Base subselect interface class */
 class select_subselect :public select_result_interceptor
 {
diff --git a/sql/sql_derived.cc b/sql/sql_derived.cc
index a910ed6290f..ce02333dfe1 100644
--- a/sql/sql_derived.cc
+++ b/sql/sql_derived.cc
@@ -788,7 +788,7 @@ bool mysql_derived_optimize(THD *thd, LEX *lex, TABLE_LIST *derived)
     if (!derived->is_merged_derived())
     {
       JOIN *join= first_select->join;
-      unit->set_limit(unit->global_parameters);
+      unit->set_limit(unit->global_parameters());
       unit->optimized= TRUE;
       if ((res= join->optimize()))
         goto err;
@@ -905,7 +905,7 @@ bool mysql_derived_fill(THD *thd, LEX *lex, TABLE_LIST *derived)
   }
   else
   {
-    unit->set_limit(unit->global_parameters);
+    unit->set_limit(unit->global_parameters());
     if (unit->select_limit_cnt == HA_POS_ERROR)
       first_select->options&= ~OPTION_FOUND_ROWS;
 
diff --git a/sql/sql_explain.cc b/sql/sql_explain.cc
index 53ac095d1d0..7ab6413bfef 100644
--- a/sql/sql_explain.cc
+++ b/sql/sql_explain.cc
@@ -229,6 +229,9 @@ int Explain_union::print_explain(Explain_query *query,
     sel->print_explain(query, output, explain_flags, is_analyze);
   }
 
+  if (!using_tmp)
+    return 0;
+
   /* Print a line with "UNION RESULT" */
   List<Item> item_list;
   Item *item_null= new Item_null();
diff --git a/sql/sql_explain.h b/sql/sql_explain.h
index a2b4ea282b7..c4de08f6a4c 100644
--- a/sql/sql_explain.h
+++ b/sql/sql_explain.h
@@ -225,6 +225,7 @@ public:
 
   const char *fake_select_type;
   bool using_filesort;
+  bool using_tmp;
 
   Table_access_tracker *get_fake_select_lex_tracker()
   {
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index 45045ff54a0..d6249f4ab21 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -470,7 +470,7 @@ void lex_start(THD *thd)
   lex->unit.next= lex->unit.master=
     lex->unit.link_next= lex->unit.return_to= 0;
   lex->unit.prev= lex->unit.link_prev= 0;
-  lex->unit.slave= lex->unit.global_parameters= lex->current_select=
+  lex->unit.slave= lex->current_select=
     lex->all_selects_list= &lex->select_lex;
   lex->select_lex.master= &lex->unit;
   lex->select_lex.prev= &lex->unit.slave;
@@ -1819,7 +1819,6 @@ void st_select_lex_unit::init_query()
 {
   st_select_lex_node::init_query();
   linkage= GLOBAL_OPTIONS_TYPE;
-  global_parameters= first_select();
   select_limit_cnt= HA_POS_ERROR;
   offset_limit_cnt= 0;
   union_distinct= 0;
@@ -1828,6 +1827,7 @@ void st_select_lex_unit::init_query()
   union_result= 0;
   table= 0;
   fake_select_lex= 0;
+  saved_fake_select_lex= 0;
   cleaned= 0;
   item_list.empty();
   describe= 0;
@@ -1901,7 +1901,6 @@ void st_select_lex::init_select()
   in_sum_expr= with_wild= 0;
   options= 0;
   sql_cache= SQL_CACHE_UNSPECIFIED;
-  braces= 0;
   interval_list.empty();
   ftfunc_list_alloc.empty();
   inner_sum_func_list= 0;
@@ -2214,6 +2213,7 @@ bool st_select_lex::test_limit()
 }
 
 
+
 st_select_lex_unit* st_select_lex_unit::master_unit()
 {
     return this;
@@ -2226,6 +2226,75 @@ st_select_lex* st_select_lex_unit::outer_select()
 }
 
 
+ha_rows st_select_lex::get_offset()
+{
+  ulonglong val= 0;
+
+  if (offset_limit)
+  {
+    // see comment for st_select_lex::get_limit()
+    bool fix_fields_successful= true;
+    if (!offset_limit->fixed)
+    {
+      fix_fields_successful= !offset_limit->fix_fields(master_unit()->thd,
+                                                       NULL);
+
+      DBUG_ASSERT(fix_fields_successful);
+    }
+    val= fix_fields_successful ? offset_limit->val_uint() : HA_POS_ERROR;
+  }
+
+  return (ha_rows)val;
+}
+
+
+ha_rows st_select_lex::get_limit()
+{
+  ulonglong val= HA_POS_ERROR;
+
+  if (select_limit)
+  {
+    /*
+      fix_fields() has not been called for select_limit. That's due to the
+      historical reasons -- this item could be only of type Item_int, and
+      Item_int does not require fix_fields(). Thus, fix_fields() was never
+      called for select_limit.
+
+      Some time ago, Item_splocal was also allowed for LIMIT / OFFSET clauses.
+      However, the fix_fields() behavior was not updated, which led to a crash
+      in some cases.
+
+      There is no single place where to call fix_fields() for LIMIT / OFFSET
+      items during the fix-fields-phase. Thus, for the sake of readability,
+      it was decided to do it here, on the evaluation phase (which is a
+      violation of design, but we chose the lesser of two evils).
+
+      We can call fix_fields() here, because select_limit can be of two
+      types only: Item_int and Item_splocal. Item_int::fix_fields() is trivial,
+      and Item_splocal::fix_fields() (or rather Item_sp_variable::fix_fields())
+      has the following properties:
+        1) it does not affect other items;
+        2) it does not fail.
+
+      Nevertheless DBUG_ASSERT was added to catch future changes in
+      fix_fields() implementation. Also added runtime check against a result
+      of fix_fields() in order to handle error condition in non-debug build.
+    */
+    bool fix_fields_successful= true;
+    if (!select_limit->fixed)
+    {
+      fix_fields_successful= !select_limit->fix_fields(master_unit()->thd,
+                                                       NULL);
+
+      DBUG_ASSERT(fix_fields_successful);
+    }
+    val= fix_fields_successful ? select_limit->val_uint() : HA_POS_ERROR;
+  }
+
+  return (ha_rows)val;
+}
+
+
 bool st_select_lex::add_order_to_list(THD *thd, Item *item, bool asc)
 {
   return add_to_list(thd, order_list, item, asc);
@@ -2237,6 +2306,7 @@ bool st_select_lex::add_gorder_to_list(THD *thd, Item *item, bool asc)
   return add_to_list(thd, gorder_list, item, asc);
 }
 
+
 bool st_select_lex::add_item_to_list(THD *thd, Item *item)
 {
   DBUG_ENTER("st_select_lex::add_item_to_list");
@@ -2364,7 +2434,7 @@ void st_select_lex_unit::print(String *str, enum_query_type query_type)
     if (sl->braces)
       str->append(')');
   }
-  if (fake_select_lex == global_parameters)
+  if (fake_select_lex)
   {
     if (fake_select_lex->order_list.elements)
     {
@@ -2375,6 +2445,8 @@ void st_select_lex_unit::print(String *str, enum_query_type query_type)
     }
     fake_select_lex->print_limit(thd, str, query_type);
   }
+  else if (saved_fake_select_lex)
+    saved_fake_select_lex->print_limit(thd, str, query_type);
 }
 
 
@@ -2425,7 +2497,7 @@ void st_select_lex::print_limit(THD *thd,
   SELECT_LEX_UNIT *unit= master_unit();
   Item_subselect *item= unit->item;
 
-  if (item && unit->global_parameters == this)
+  if (item && unit->global_parameters() == this)
   {
     Item_subselect::subs_type subs_type= item->substype();
     if (subs_type == Item_subselect::EXISTS_SUBS ||
@@ -2817,97 +2889,39 @@ LEX::copy_db_to(char **p_db, size_t *p_db_length) const
   return thd->copy_db_to(p_db, p_db_length);
 }
 
-/*
-  initialize limit counters
+/**
+  Initialize offset and limit counters.
 
-  SYNOPSIS
-    st_select_lex_unit::set_limit()
-    values	- SELECT_LEX with initial values for counters
+  @param sl SELECT_LEX to get offset and limit from.
 */
 
 void st_select_lex_unit::set_limit(st_select_lex *sl)
 {
-  ha_rows select_limit_val;
-  ulonglong val;
+  DBUG_ASSERT(!thd->stmt_arena->is_stmt_prepare());
 
-  DBUG_ASSERT(! thd->stmt_arena->is_stmt_prepare());
-  if (sl->select_limit)
-  {
-    Item *item = sl->select_limit;
-    /*
-      fix_fields() has not been called for sl->select_limit. That's due to the
-      historical reasons -- this item could be only of type Item_int, and
-      Item_int does not require fix_fields(). Thus, fix_fields() was never
-      called for sl->select_limit.
-
-      Some time ago, Item_splocal was also allowed for LIMIT / OFFSET clauses.
-      However, the fix_fields() behavior was not updated, which led to a crash
-      in some cases.
-
-      There is no single place where to call fix_fields() for LIMIT / OFFSET
-      items during the fix-fields-phase. Thus, for the sake of readability,
-      it was decided to do it here, on the evaluation phase (which is a
-      violation of design, but we chose the lesser of two evils).
-
-      We can call fix_fields() here, because sl->select_limit can be of two
-      types only: Item_int and Item_splocal. Item_int::fix_fields() is trivial,
-      and Item_splocal::fix_fields() (or rather Item_sp_variable::fix_fields())
-      has the following specific:
-        1) it does not affect other items;
-        2) it does not fail.
-
-      Nevertheless DBUG_ASSERT was added to catch future changes in
-      fix_fields() implementation. Also added runtime check against a result
-      of fix_fields() in order to handle error condition in non-debug build.
-    */
-    bool fix_fields_successful= true;
-    if (!item->fixed)
-    {
-      fix_fields_successful= !item->fix_fields(thd, NULL);
-
-      DBUG_ASSERT(fix_fields_successful);
-    }
-    val= fix_fields_successful ? item->val_uint() : HA_POS_ERROR;
-  }
+  offset_limit_cnt= sl->get_offset();
+  select_limit_cnt= sl->get_limit();
+  if (select_limit_cnt + offset_limit_cnt >= select_limit_cnt)
+    select_limit_cnt+= offset_limit_cnt;
   else
-    val= HA_POS_ERROR;
+    select_limit_cnt= HA_POS_ERROR;
+}
 
-  select_limit_val= (ha_rows)val;
-#ifndef BIG_TABLES
-  /*
-    Check for overflow : ha_rows can be smaller then ulonglong if
-    BIG_TABLES is off.
-    */
-  if (val != (ulonglong)select_limit_val)
-    select_limit_val= HA_POS_ERROR;
-#endif
-  if (sl->offset_limit)
-  {
-    Item *item = sl->offset_limit;
-    // see comment for sl->select_limit branch.
-    bool fix_fields_successful= true;
-    if (!item->fixed)
-    {
-      fix_fields_successful= !item->fix_fields(thd, NULL);
 
-      DBUG_ASSERT(fix_fields_successful);
-    }
-    val= fix_fields_successful ? item->val_uint() : 0;
-  }
-  else
-    val= 0;
+/**
+  Decide if a temporary table is needed for the UNION.
 
-  offset_limit_cnt= (ha_rows)val;
-#ifndef BIG_TABLES
-  /* Check for truncation. */
-  if (val != (ulonglong)offset_limit_cnt)
-    offset_limit_cnt= HA_POS_ERROR;
-#endif
-  select_limit_cnt= select_limit_val + offset_limit_cnt;
-  if (select_limit_cnt < select_limit_val)
-    select_limit_cnt= HA_POS_ERROR;		// no limit
-}
+  @retval true  A temporary table is needed.
+  @retval false A temporary table is not needed.
+ */
 
+bool st_select_lex_unit::union_needs_tmp_table()
+{
+  return union_distinct != NULL ||
+    global_parameters()->order_list.elements != 0 ||
+    thd->lex->sql_command == SQLCOM_INSERT_SELECT ||
+    thd->lex->sql_command == SQLCOM_REPLACE_SELECT;
+}  
 
 /**
   @brief Set the initial purpose of this TABLE_LIST object in the list of used
@@ -3496,7 +3510,7 @@ bool st_select_lex::optimize_unflattened_subqueries(bool const_only)
         ulonglong save_options;
         int res;
         /* We need only 1 row to determine existence */
-        un->set_limit(un->global_parameters);
+        un->set_limit(un->global_parameters());
         un->thd->lex->current_select= sl;
         save_options= inner_join->select_options;
         if (options & SELECT_DESCRIBE)
@@ -3896,7 +3910,7 @@ void SELECT_LEX::update_used_tables()
   }
   for (ORDER *order= group_list.first; order; order= order->next)
     (*order->item)->update_used_tables();
-  if (!master_unit()->is_union() || master_unit()->global_parameters != this)
+  if (!master_unit()->is_union() || master_unit()->global_parameters() != this)
   {
     for (ORDER *order= order_list.first; order; order= order->next)
       (*order->item)->update_used_tables();
@@ -4244,7 +4258,8 @@ int st_select_lex_unit::save_union_explain(Explain_query *output)
     eu->add_select(sl->select_number);
 
   eu->fake_select_type= "UNION RESULT";
-  eu->using_filesort= MY_TEST(global_parameters->order_list.first);
+  eu->using_filesort= MY_TEST(global_parameters()->order_list.first);
+  eu->using_tmp= union_needs_tmp_table();
 
   // Save the UNION node
   output->add_node(eu);
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index 5e9c7b9dc6a..dd43167511f 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -577,11 +577,28 @@ public:
     any SELECT of this unit execution
   */
   List<Item> types;
-  /*
-    Pointer to 'last' select or pointer to unit where stored
-    global parameters for union
+  /**
+    Pointer to 'last' select, or pointer to select where we stored
+    global parameters for union.
+
+    If this is a union of multiple selects, the parser puts the global
+    parameters in fake_select_lex. If the union doesn't use a
+    temporary table, st_select_lex_unit::prepare() nulls out
+    fake_select_lex, but saves a copy in saved_fake_select_lex in
+    order to preserve the global parameters.
+
+    If it is not a union, first_select() is the last select.
+
+    @return select containing the global parameters
   */
-  st_select_lex *global_parameters;
+  inline st_select_lex *global_parameters()
+  {
+    if (fake_select_lex != NULL)
+      return fake_select_lex;
+    else if (saved_fake_select_lex != NULL)
+      return saved_fake_select_lex;
+    return first_select();
+  };
   //node on wich we should return current_select pointer after parsing subquery
   st_select_lex *return_to;
   /* LIMIT clause runtime counters */
@@ -600,6 +617,11 @@ public:
     ORDER BY and LIMIT
   */
   st_select_lex *fake_select_lex;
+  /**
+    SELECT_LEX that stores LIMIT and OFFSET for UNION ALL when no
+    fake_select_lex is used.
+  */
+  st_select_lex *saved_fake_select_lex;
 
   st_select_lex *union_distinct; /* pointer to the last UNION DISTINCT */
   bool describe; /* union exec() called for EXPLAIN */
@@ -645,6 +667,7 @@ public:
   void set_limit(st_select_lex *values);
   void set_thd(THD *thd_arg) { thd= thd_arg; }
   inline bool is_union (); 
+  bool union_needs_tmp_table();
 
   void set_unique_exclude();
 
@@ -904,9 +927,26 @@ public:
   */
   void cut_subtree() { slave= 0; }
   bool test_limit();
+  /**
+    Get offset for LIMIT.
+
+    Evaluate offset item if necessary.
+
+    @return Number of rows to skip.
+  */
+  ha_rows get_offset();
+  /**
+   Get limit.
+
+   Evaluate limit item if necessary.
+
+   @return Limit of rows in result.
+  */
+  ha_rows get_limit();
 
   friend void lex_start(THD *thd);
-  st_select_lex() : group_list_ptrs(NULL), n_sum_items(0), n_child_sum_items(0)
+  st_select_lex() : group_list_ptrs(NULL), braces(0),
+  n_sum_items(0), n_child_sum_items(0)
   {}
   void make_empty_select()
   {
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index d8906b2d578..4f9f9339599 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -5592,7 +5592,7 @@ static bool execute_sqlcom_select(THD *thd, TABLE_LIST *all_tables)
   bool res;
   /* assign global limit variable if limit is not given */
   {
-    SELECT_LEX *param= lex->unit.global_parameters;
+    SELECT_LEX *param= lex->unit.global_parameters();
     if (!param->explicit_limit)
       param->select_limit=
         new Item_int((ulonglong) thd->variables.select_limit);
@@ -6675,7 +6675,6 @@ mysql_new_select(LEX *lex, bool move_down)
                 unit->first_select()->context.outer_context;
   }
 
-  select_lex->master_unit()->global_parameters= select_lex;
   select_lex->include_global((st_select_lex_node**)&lex->all_selects_list);
   lex->current_select= select_lex;
   /*
@@ -7642,7 +7641,7 @@ bool st_select_lex_unit::add_fake_select_lex(THD *thd_arg)
   fake_select_lex->context.outer_context=first_sl->context.outer_context;
   /* allow item list resolving in fake select for ORDER BY */
   fake_select_lex->context.resolve_in_select_list= TRUE;
-  fake_select_lex->context.select_lex= fake_select_lex;
+  fake_select_lex->context.select_lex= fake_select_lex;  
 
   if (!is_union())
   {
@@ -7652,7 +7651,6 @@ bool st_select_lex_unit::add_fake_select_lex(THD *thd_arg)
       (SELECT ... LIMIT n) ORDER BY order_list [LIMIT m]
       just before the parser starts processing order_list
     */ 
-    global_parameters= fake_select_lex;
     fake_select_lex->no_table_names_allowed= 1;
     thd_arg->lex->current_select= fake_select_lex;
   }
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 436a2f6ce5d..a3d4d32bca3 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -351,7 +351,7 @@ bool handle_select(THD *thd, LEX *lex, select_result *result,
   else
   {
     SELECT_LEX_UNIT *unit= &lex->unit;
-    unit->set_limit(unit->global_parameters);
+    unit->set_limit(unit->global_parameters());
     /*
       'options' of mysql_select will be set in JOIN, as far as JOIN for
       every PS/SP execution new, we will not need reset this flag if 
@@ -792,7 +792,8 @@ JOIN::prepare(Item ***rref_pointer_array,
   ref_pointer_array= *rref_pointer_array;
 
   /* Resolve the ORDER BY that was skipped, then remove it. */
-  if (skip_order_by && select_lex != select_lex->master_unit()->global_parameters)
+  if (skip_order_by && select_lex !=
+                       select_lex->master_unit()->global_parameters())
   {
     if (setup_order(thd, (*rref_pointer_array), tables_list, fields_list,
                     all_fields, select_lex->order_list.first))
@@ -2481,13 +2482,6 @@ void JOIN::exec_inner()
     thd->set_examined_row_count(0);
     DBUG_VOID_RETURN;
   }
-  /*
-    Don't reset the found rows count if there're no tables as
-    FOUND_ROWS() may be called. Never reset the examined row count here.
-    It must be accumulated from all join iterations of all join parts.
-  */
-  if (table_count)
-    thd->limit_found_rows= 0;
 
   /*
     Evaluate expensive constant conditions that were not evaluated during
@@ -3091,7 +3085,6 @@ void JOIN::exec_inner()
                                     *curr_fields_list),
                                    Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF);
   error= do_select(curr_join, curr_fields_list, NULL, procedure);
-  thd->limit_found_rows= curr_join->send_records;
   if (curr_join->order && curr_join->filesort_found_rows)
   {
     /* Use info provided by filesort. */
@@ -3257,7 +3250,7 @@ mysql_select(THD *thd, Item ***rref_pointer_array,
       if (select_lex->linkage != GLOBAL_OPTIONS_TYPE)
       {
 	//here is EXPLAIN of subselect or derived table
-	if (join->change_result(result))
+	if (join->change_result(result, NULL))
 	{
 	  DBUG_RETURN(TRUE);
 	}
@@ -12088,6 +12081,14 @@ return_zero_rows(JOIN *join, select_result *result, List<TABLE_LIST> &tables,
     if (having && having->val_int() == 0)
       send_row=0;
   }
+
+  /* Update results for FOUND_ROWS */
+  if (!join->send_row_on_empty_set())
+  {
+    join->thd->set_examined_row_count(0);
+    join->thd->limit_found_rows= 0;
+  }
+
   if (!(result->send_result_set_metadata(fields,
                               Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF)))
   {
@@ -12097,9 +12098,6 @@ return_zero_rows(JOIN *join, select_result *result, List<TABLE_LIST> &tables,
     if (!send_error)
       result->send_eof();				// Should be safe
   }
-  /* Update results for FOUND_ROWS */
-  join->thd->limit_found_rows= 0;
-  join->thd->set_examined_row_count(0);
   DBUG_RETURN(0);
 }
 
@@ -17300,6 +17298,9 @@ do_select(JOIN *join,List<Item> *fields,TABLE *table,Procedure *procedure)
     if (error == NESTED_LOOP_QUERY_LIMIT)
       error= NESTED_LOOP_OK;                    /* select_limit used */
   }
+
+  join->thd->limit_found_rows= join->send_records;
+
   if (error == NESTED_LOOP_NO_MORE_ROWS || join->thd->killed == ABORT_QUERY)
     error= NESTED_LOOP_OK;
 
@@ -23257,7 +23258,7 @@ int print_fake_select_lex_join(select_result_sink *result, bool on_the_fly,
   /* rows */
   item_list.push_back(item_null);
   /* extra */
-  if (select_lex->master_unit()->global_parameters->order_list.first)
+  if (select_lex->master_unit()->global_parameters()->order_list.first)
     item_list.push_back(new Item_string("Using filesort",
                                         14, cs));
   else
@@ -23934,16 +23935,19 @@ bool mysql_explain_union(THD *thd, SELECT_LEX_UNIT *unit, select_result *result)
 
   if (unit->is_union())
   {
-    unit->fake_select_lex->select_number= FAKE_SELECT_LEX_ID; // jost for initialization
-    unit->fake_select_lex->type= "UNION RESULT";
-    unit->fake_select_lex->options|= SELECT_DESCRIBE;
+    if (unit->union_needs_tmp_table())
+    {
+      unit->fake_select_lex->select_number= FAKE_SELECT_LEX_ID; // just for initialization
+      unit->fake_select_lex->type= "UNION RESULT";
+      unit->fake_select_lex->options|= SELECT_DESCRIBE;
+    }
     if (!(res= unit->prepare(thd, result, SELECT_NO_UNLOCK | SELECT_DESCRIBE)))
       res= unit->exec();
   }
   else
   {
     thd->lex->current_select= first;
-    unit->set_limit(unit->global_parameters);
+    unit->set_limit(unit->global_parameters());
     res= mysql_select(thd, &first->ref_pointer_array,
 			first->table_list.first,
 			first->with_wild, first->item_list,
@@ -24427,28 +24431,34 @@ void st_select_lex::print(THD *thd, String *str, enum_query_type query_type)
 
 
 /**
-  change select_result object of JOIN.
+  Change the select_result object of the JOIN.
 
-  @param res		new select_result object
+  If old_result is not used, forward the call to the current
+  select_result in case it is a wrapper around old_result.
 
-  @retval
-    FALSE   OK
-  @retval
-    TRUE    error
+  Call prepare() and prepare2() on the new select_result if we decide
+  to use it.
+
+  @param new_result New select_result object
+  @param old_result Old select_result object (NULL to force change)
+
+  @retval false Success
+  @retval true  Error
 */
 
-bool JOIN::change_result(select_result *res)
+bool JOIN::change_result(select_result *new_result, select_result *old_result)
 {
   DBUG_ENTER("JOIN::change_result");
-  result= res;
-  if (tmp_join)
-    tmp_join->result= res;
-  if (!procedure && (result->prepare(fields_list, select_lex->master_unit()) ||
-                     result->prepare2()))
+  if (old_result == NULL || result == old_result)
   {
-    DBUG_RETURN(TRUE);
+    result= new_result;
+    if (result->prepare(fields_list, select_lex->master_unit()) ||
+        result->prepare2())
+      DBUG_RETURN(true); /* purecov: inspected */
+    DBUG_RETURN(false);
   }
-  DBUG_RETURN(FALSE);
+  else
+    DBUG_RETURN(result->change_result(new_result));
 }
 
 
diff --git a/sql/sql_select.h b/sql/sql_select.h
index 63fd6a6d99f..01f3c7834a8 100644
--- a/sql/sql_select.h
+++ b/sql/sql_select.h
@@ -1413,7 +1413,7 @@ public:
             having_value != Item::COND_FALSE);
   }
   bool empty_result() { return (zero_result_cause && !implicit_grouping); }
-  bool change_result(select_result *result);
+  bool change_result(select_result *new_result, select_result *old_result);
   bool is_top_level_join() const
   {
     return (unit == &thd->lex->unit && (unit->fake_select_lex == 0 ||
diff --git a/sql/sql_union.cc b/sql/sql_union.cc
index 9d068e464f5..cfc34b0f90a 100644
--- a/sql/sql_union.cc
+++ b/sql/sql_union.cc
@@ -187,6 +187,103 @@ void select_union::cleanup()
 }
 
 
+
+/**
+  Replace the current result with new_result and prepare it.  
+
+  @param new_result New result pointer
+
+  @retval FALSE Success
+  @retval TRUE  Error
+*/
+
+bool select_union_direct::change_result(select_result *new_result)
+{
+  result= new_result;
+  return (result->prepare(unit->types, unit) || result->prepare2());
+}
+
+
+bool select_union_direct::postponed_prepare(List<Item> &types)
+{
+  if (result != NULL)
+    return (result->prepare(types, unit) || result->prepare2());
+  else
+    return false;
+}
+
+
+bool select_union_direct::send_result_set_metadata(List<Item> &list, uint flags)
+{
+  if (done_send_result_set_metadata)
+    return false;
+  done_send_result_set_metadata= true;
+
+  /*
+    Set global offset and limit to be used in send_data(). These can
+    be variables in prepared statements or stored programs, so they
+    must be reevaluated for each execution.
+   */
+  offset= unit->global_parameters()->get_offset();
+  limit= unit->global_parameters()->get_limit();
+  if (limit + offset >= limit)
+    limit+= offset;
+  else
+    limit= HA_POS_ERROR; /* purecov: inspected */
+
+  return result->send_result_set_metadata(unit->types, flags);
+}
+
+
+int select_union_direct::send_data(List<Item> &items)
+{
+  if (!limit)
+    return false;
+  limit--;
+  if (offset)
+  {
+    offset--;
+    return false;
+  }
+
+  fill_record(thd, table, table->field, items, true, false);
+  if (thd->is_error())
+    return true; /* purecov: inspected */
+
+  return result->send_data(unit->item_list);
+}
+
+
+bool select_union_direct::initialize_tables (JOIN *join)
+{
+  if (done_initialize_tables)
+    return false;
+  done_initialize_tables= true;
+
+  return result->initialize_tables(join);
+}
+
+
+bool select_union_direct::send_eof()
+{
+  // Reset for each SELECT_LEX, so accumulate here
+  limit_found_rows+= thd->limit_found_rows;
+
+  if (unit->thd->lex->current_select == last_select_lex)
+  {
+    thd->limit_found_rows= limit_found_rows;
+
+    // Reset and make ready for re-execution
+    done_send_result_set_metadata= false;
+    done_initialize_tables= false;
+
+    return result->send_eof();
+  }
+  else
+    return false;
+}
+
+
 /*
   initialization procedures before fake_select_lex preparation()
 
@@ -217,12 +314,12 @@ st_select_lex_unit::init_prepare_fake_select_lex(THD *thd_arg,
   */    
   if (!fake_select_lex->first_execution && first_execution)
   {
-    for (ORDER *order= global_parameters->order_list.first;
+    for (ORDER *order= global_parameters()->order_list.first;
          order;
          order= order->next)
       order->item= &order->item_ptr;
   }
-  for (ORDER *order= global_parameters->order_list.first;
+  for (ORDER *order= global_parameters()->order_list.first;
        order;
        order=order->next)
   {
@@ -241,11 +338,19 @@ bool st_select_lex_unit::prepare(THD *thd_arg, select_result *sel_result,
   SELECT_LEX *sl, *first_sl= first_select();
   select_result *tmp_result;
   bool is_union_select;
+  bool instantiate_tmp_table= false;
   DBUG_ENTER("st_select_lex_unit::prepare");
 
   describe= MY_TEST(additional_options & SELECT_DESCRIBE);
 
   /*
+    Save fake_select_lex in case we don't need it for anything but
+    global parameters.
+  */
+  if (saved_fake_select_lex == NULL) // Don't overwrite on PS second prepare
+    saved_fake_select_lex= fake_select_lex;
+
+  /*
     result object should be reassigned even if preparing already done for
     max/min subquery (ALL/ANY optimization)
   */
@@ -283,10 +388,22 @@ bool st_select_lex_unit::prepare(THD *thd_arg, select_result *sel_result,
 
   if (is_union_select)
   {
-    if (!(tmp_result= union_result= new select_union))
-      goto err;
-    if (describe)
-      tmp_result= sel_result;
+    if (is_union() && !union_needs_tmp_table())
+    {
+      SELECT_LEX *last= first_select();
+      while (last->next_select())
+        last= last->next_select();
+      if (!(tmp_result= union_result= new select_union_direct(sel_result, last)))
+        goto err; /* purecov: inspected */
+      fake_select_lex= NULL;
+      instantiate_tmp_table= false;
+    }
+    else
+    {
+      if (!(tmp_result= union_result= new select_union()))
+        goto err; /* purecov: inspected */
+      instantiate_tmp_table= true;
+    }
   }
   else
     tmp_result= sel_result;
@@ -387,6 +504,14 @@ bool st_select_lex_unit::prepare(THD *thd_arg, select_result *sel_result,
     }
   }
 
+  /*
+    If the query is using select_union_direct, we have postponed
+    preparation of the underlying select_result until column types
+    are known.
+  */
+  if (union_result != NULL && union_result->postponed_prepare(types))
+    DBUG_RETURN(true);
+
   if (is_union_select)
   {
     /*
@@ -423,13 +548,13 @@ bool st_select_lex_unit::prepare(THD *thd_arg, select_result *sel_result,
       the meaning of these accumulated flags and what to carry over to the
       recipient query (SELECT_LEX).
     */
-    if (global_parameters->ftfunc_list->elements && 
-        global_parameters->order_list.elements &&
-        global_parameters != fake_select_lex)
+    if (global_parameters()->ftfunc_list->elements && 
+        global_parameters()->order_list.elements &&
+        global_parameters() != fake_select_lex)
     {
       ORDER *ord;
       Item_func::Functype ft=  Item_func::FT_FUNC;
-      for (ord= global_parameters->order_list.first; ord; ord= ord->next)
+      for (ord= global_parameters()->order_list.first; ord; ord= ord->next)
         if ((*ord->item)->walk (&Item::find_function_processor, FALSE, 
                                 (uchar *) &ft))
         {
@@ -447,11 +572,12 @@ bool st_select_lex_unit::prepare(THD *thd_arg, select_result *sel_result,
       from it (this should be removed in 5.2 when fulltext search is moved 
       out of MyISAM).
     */
-    if (global_parameters->ftfunc_list->elements)
+    if (global_parameters()->ftfunc_list->elements)
       create_options= create_options | TMP_TABLE_FORCE_MYISAM;
 
     if (union_result->create_result_table(thd, &types, MY_TEST(union_distinct),
-                                          create_options, "", FALSE, TRUE))
+                                          create_options, "", false,
+                                          instantiate_tmp_table))
       goto err;
     if (fake_select_lex && !fake_select_lex->first_cond_optimization)
     {
@@ -485,7 +611,7 @@ bool st_select_lex_unit::prepare(THD *thd_arg, select_result *sel_result,
       if (saved_error)
         goto err;
 
-      if (thd->stmt_arena->is_stmt_prepare())
+      if (fake_select_lex != NULL && thd->stmt_arena->is_stmt_prepare())
       {
         /* Validate the global parameters of this union */
 
@@ -511,14 +637,14 @@ bool st_select_lex_unit::prepare(THD *thd_arg, select_result *sel_result,
           We need to add up n_sum_items in order to make the correct
           allocation in setup_ref_array().
         */
-        fake_select_lex->n_child_sum_items+= global_parameters->n_sum_items;
+        fake_select_lex->n_child_sum_items+= global_parameters()->n_sum_items;
 
 	saved_error= fake_select_lex->join->
 	  prepare(&fake_select_lex->ref_pointer_array,
 		  fake_select_lex->table_list.first,
 		  0, 0,
-                  global_parameters->order_list.elements, // og_num
-                  global_parameters->order_list.first,    // order
+                  global_parameters()->order_list.elements, // og_num
+                  global_parameters()->order_list.first,    // order
                   false, NULL, NULL, NULL,
 		  fake_select_lex, this);
 	fake_select_lex->table_list.empty();
@@ -570,7 +696,11 @@ bool st_select_lex_unit::optimize()
       {
         item->assigned(0); // We will reinit & rexecute unit
         item->reset();
-        table->file->ha_delete_all_rows();
+        if (table->created)
+        {
+          table->file->ha_delete_all_rows();
+          table->file->info(HA_STATUS_VARIABLE);
+        }
       }
       /* re-enabling indexes for next subselect iteration */
       if (union_distinct && table->file->ha_enable_indexes(HA_KEY_SWITCH_ALL))
@@ -587,7 +717,7 @@ bool st_select_lex_unit::optimize()
       else
       {
         set_limit(sl);
-	if (sl == global_parameters || describe)
+	if (sl == global_parameters() || describe)
 	{
 	  offset_limit_cnt= 0;
 	  /*
@@ -656,14 +786,17 @@ bool st_select_lex_unit::exec()
     {
       ha_rows records_at_start= 0;
       thd->lex->current_select= sl;
-      if (sl != &thd->lex->select_lex)
-        fake_select_lex->uncacheable|= sl->uncacheable;
-      else
-        fake_select_lex->uncacheable= 0;
+      if (fake_select_lex)
+      {
+        if (sl != &thd->lex->select_lex)
+          fake_select_lex->uncacheable|= sl->uncacheable;
+        else
+          fake_select_lex->uncacheable= 0;
+      }
 
       {
         set_limit(sl);
-	if (sl == global_parameters || describe)
+	if (sl == global_parameters() || describe)
 	{
 	  offset_limit_cnt= 0;
 	  /*
@@ -690,6 +823,8 @@ bool st_select_lex_unit::exec()
 	sl->join->exec();
         if (sl == union_distinct)
 	{
+          // This is UNION DISTINCT, so there should be a fake_select_lex
+          DBUG_ASSERT(fake_select_lex != NULL);
 	  if (table->file->ha_disable_indexes(HA_KEY_SWITCH_ALL))
 	    DBUG_RETURN(TRUE);
 	  table->no_keyread=1;
@@ -714,12 +849,15 @@ bool st_select_lex_unit::exec()
 	thd->lex->current_select= lex_select_save;
 	DBUG_RETURN(saved_error);
       }
-      /* Needed for the following test and for records_at_start in next loop */
-      int error= table->file->info(HA_STATUS_VARIABLE);
-      if(error)
+      if (fake_select_lex != NULL)
       {
-        table->file->print_error(error, MYF(0));
-        DBUG_RETURN(1);
+        /* Needed for the following test and for records_at_start in next loop */
+        int error= table->file->info(HA_STATUS_VARIABLE);
+        if(error)
+        {
+          table->file->print_error(error, MYF(0));
+          DBUG_RETURN(1);
+        }
       }
       if (found_rows_for_union && !sl->braces && 
           select_limit_cnt != HA_POS_ERROR)
@@ -752,8 +890,6 @@ bool st_select_lex_unit::exec()
 
   DBUG_EXECUTE_IF("show_explain_probe_union_read", 
                    dbug_serve_apcs(thd, 1););
-  /* Send result to 'result' */
-  saved_error= TRUE;
   {
     List<Item_func_match> empty_list;
     empty_list.empty();
@@ -763,11 +899,15 @@ bool st_select_lex_unit::exec()
     */
     thd->lex->limit_rows_examined_cnt= ULONGLONG_MAX;
 
-    if (!thd->is_fatal_error)				// Check if EOM
+    if (fake_select_lex != NULL && !thd->is_fatal_error)    // Check if EOM
     {
-      set_limit(global_parameters);
+       /* Send result to 'result' */
+       saved_error= true;
+
+      set_limit(global_parameters());
       init_prepare_fake_select_lex(thd, first_execution);
       JOIN *join= fake_select_lex->join;
+      saved_error= false;
       if (!join)
       {
 	/*
@@ -799,7 +939,7 @@ bool st_select_lex_unit::exec()
           for this (with a different join object)
         */
         if (!fake_select_lex->ref_pointer_array)
-          fake_select_lex->n_child_sum_items+= global_parameters->n_sum_items;
+          fake_select_lex->n_child_sum_items+= global_parameters()->n_sum_items;
         
         if (!was_executed)
           save_union_explain_part2(thd->lex->explain);
@@ -807,8 +947,8 @@ bool st_select_lex_unit::exec()
         saved_error= mysql_select(thd, &fake_select_lex->ref_pointer_array,
                               &result_table_list,
                               0, item_list, NULL,
-                              global_parameters->order_list.elements,
-                              global_parameters->order_list.first,
+				  global_parameters()->order_list.elements,
+				  global_parameters()->order_list.first,
                               NULL, NULL, NULL,
                               fake_select_lex->options | SELECT_NO_UNLOCK,
                               result, this, fake_select_lex);
@@ -830,8 +970,8 @@ bool st_select_lex_unit::exec()
           saved_error= mysql_select(thd, &fake_select_lex->ref_pointer_array,
                                 &result_table_list,
                                 0, item_list, NULL,
-                                global_parameters->order_list.elements,
-                                global_parameters->order_list.first,
+				    global_parameters()->order_list.elements,
+				    global_parameters()->order_list.first,
                                 NULL, NULL, NULL,
                                 fake_select_lex->options | SELECT_NO_UNLOCK,
                                 result, this, fake_select_lex);
@@ -907,11 +1047,11 @@ bool st_select_lex_unit::cleanup()
       Note: global_parameters and fake_select_lex are always
             initialized for UNION
     */
-    DBUG_ASSERT(global_parameters);
-    if (global_parameters->order_list.elements)
+    DBUG_ASSERT(global_parameters());
+    if (global_parameters()->order_list.elements)
     {
       ORDER *ord;
-      for (ord= global_parameters->order_list.first; ord; ord= ord->next)
+      for (ord= global_parameters()->order_list.first; ord; ord= ord->next)
         (*ord->item)->walk (&Item::cleanup_processor, 0, 0);
     }
   }
@@ -942,32 +1082,33 @@ void st_select_lex_unit::reinit_exec_mechanism()
 }
 
 
-/*
-  change select_result object of unit
+/**
+  Change the select_result object used to return the final result of
+  the unit, replacing occurences of old_result with new_result.
 
-  SYNOPSIS
-    st_select_lex_unit::change_result()
-    result	new select_result object
-    old_result	old select_result object
+  @param new_result New select_result object
+  @param old_result Old select_result object
 
-  RETURN
-    FALSE - OK
-    TRUE  - error
+  @retval false Success
+  @retval true  Error
 */
 
 bool st_select_lex_unit::change_result(select_result_interceptor *new_result,
                                        select_result_interceptor *old_result)
 {
-  bool res= FALSE;
   for (SELECT_LEX *sl= first_select(); sl; sl= sl->next_select())
   {
-    if (sl->join && sl->join->result == old_result)
-      if (sl->join->change_result(new_result))
-	return TRUE;
+    if (sl->join)
+      if (sl->join->change_result(new_result, old_result))
+	return true; /* purecov: inspected */
   }
-  if (fake_select_lex && fake_select_lex->join)
-    res= fake_select_lex->join->change_result(new_result);
-  return (res);
+  /*
+    If there were a fake_select_lex->join, we would have to change the
+    result of that also, but change_result() is called before such an
+    object is created.
+  */
+  DBUG_ASSERT(fake_select_lex == NULL || fake_select_lex->join == NULL);
+  return false;
 }
 
 /*
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index e7fcdfbe596..0bf751e7feb 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -657,6 +657,17 @@ bool add_select_to_union_list(LEX *lex, bool is_union_distinct,
     my_error(ER_WRONG_USAGE, MYF(0), "UNION", "INTO");
     return TRUE;
   }
+  if (lex->current_select->order_list.first && !lex->current_select->braces)
+  {
+    my_error(ER_WRONG_USAGE, MYF(0), "UNION", "ORDER BY");
+    return TRUE;
+  }
+
+  if (lex->current_select->explicit_limit && !lex->current_select->braces)
+  {
+    my_error(ER_WRONG_USAGE, MYF(0), "UNION", "LIMIT");
+    return TRUE;
+  }
   if (lex->current_select->linkage == GLOBAL_OPTIONS_TYPE)
   {
     my_parse_error(ER(ER_SYNTAX_ERROR));
@@ -684,11 +695,14 @@ bool add_select_to_union_list(LEX *lex, bool is_union_distinct,
 bool setup_select_in_parentheses(LEX *lex) 
 {
   SELECT_LEX * sel= lex->current_select;
+  /*
   if (sel->set_braces(1))
   {
     my_parse_error(ER(ER_SYNTAX_ERROR));
     return TRUE;
   }
+  */
+  DBUG_ASSERT(sel->braces);
   if (sel->linkage == UNION_TYPE &&
       !sel->master_unit()->first_select()->braces &&
       sel->master_unit()->first_select()->linkage ==
@@ -704,10 +718,6 @@ bool setup_select_in_parentheses(LEX *lex)
     my_error(ER_WRONG_USAGE, MYF(0), "CUBE/ROLLUP", "ORDER BY");
     return TRUE;
   }
-  /* select in braces, can't contain global parameters */
-  if (sel->master_unit()->fake_select_lex)
-    sel->master_unit()->global_parameters=
-      sel->master_unit()->fake_select_lex;
   return FALSE;
 }
 
@@ -8309,6 +8319,13 @@ select_init:
         ;
 
 select_paren:
+          {
+            /*
+              In order to correctly parse UNION's global ORDER BY we need to
+              set braces before parsing the clause.
+            */
+            Lex->current_select->set_braces(true);
+          }
           SELECT_SYM select_part2
           {
             if (setup_select_in_parentheses(Lex))
@@ -8319,6 +8336,9 @@ select_paren:
 
 /* The equivalent of select_paren for nested queries. */
 select_paren_derived:
+          {
+            Lex->current_select->set_braces(true);
+          }
           SELECT_SYM select_part2_derived
           {
             if (setup_select_in_parentheses(Lex))
@@ -8331,18 +8351,8 @@ select_init2:
           select_part2
           {
             LEX *lex= Lex;
-            SELECT_LEX * sel= lex->current_select;
-            if (lex->current_select->set_braces(0))
-            {
-              my_parse_error(ER(ER_SYNTAX_ERROR));
-              MYSQL_YYABORT;
-            }
-            if (sel->linkage == UNION_TYPE &&
-                sel->master_unit()->first_select()->braces)
-            {
-              my_parse_error(ER(ER_SYNTAX_ERROR));
-              MYSQL_YYABORT;
-            }
+            /* Parentheses carry no meaning here */
+            lex->current_select->set_braces(false);
           }
           union_clause
         ;
@@ -8371,13 +8381,14 @@ select_into:
         ;
 
 select_from:
-          FROM join_table_list where_clause group_clause having_clause
-          opt_order_clause opt_limit_clause procedure_clause
+          FROM join_table_list 
           {
             Select->context.table_list=
               Select->context.first_name_resolution_table=
                 Select->table_list.first;
           }
+          where_clause group_clause having_clause
+          opt_order_clause opt_limit_clause procedure_clause
         | FROM DUAL_SYM where_clause opt_limit_clause
           /* oracle compatibility: oracle always requires FROM clause,
              and DUAL is system table without fields.
@@ -10698,10 +10709,6 @@ table_factor:
                 my_parse_error(ER(ER_SYNTAX_ERROR));
                 MYSQL_YYABORT;
               }
-              /* select in braces, can't contain global parameters */
-              if (sel->master_unit()->fake_select_lex)
-                sel->master_unit()->global_parameters=
-                   sel->master_unit()->fake_select_lex;
             }
             if ($2->init_nested_join(lex->thd))
               MYSQL_YYABORT;
@@ -11246,7 +11253,8 @@ order_clause:
                        "CUBE/ROLLUP", "ORDER BY");
               MYSQL_YYABORT;
             }
-            if (lex->sql_command != SQLCOM_ALTER_TABLE && !unit->fake_select_lex)
+            if (lex->sql_command != SQLCOM_ALTER_TABLE &&
+                !unit->fake_select_lex)
             {
               /*
                 A query of the of the form (SELECT ...) ORDER BY order_list is
@@ -11263,9 +11271,24 @@ order_clause:
                   unit->add_fake_select_lex(lex->thd))
                 MYSQL_YYABORT;
             }
+            if (sel->master_unit()->is_union() && !sel->braces)
+            {
+               /*
+                 At this point we don't know yet whether this is the last
+                 select in union or not, but we move ORDER BY to
+                 fake_select_lex anyway. If there would be one more select
+                 in union mysql_new_select will correctly throw error.
+               */
+               DBUG_ASSERT(sel->master_unit()->fake_select_lex);
+               lex->current_select= sel->master_unit()->fake_select_lex;
+               lex->push_context(&lex->current_select->context);
+             }
           }
           order_list
-        ;
+          {
+
+          }
+         ;
 
 order_list:
           order_list ',' order_ident order_dir
@@ -11285,6 +11308,13 @@ opt_limit_clause_init:
           {
             LEX *lex= Lex;
             SELECT_LEX *sel= lex->current_select;
+            if (sel->master_unit()->is_union() && !sel->braces)
+            {
+              /* Move LIMIT that belongs to UNION to fake_select_lex */
+              Lex->current_select= sel->master_unit()->fake_select_lex;
+              DBUG_ASSERT(Select);
+            }
+            sel= lex->current_select;
             sel->offset_limit= 0;
             sel->select_limit= 0;
 	    lex->limit_rows_examined= 0;
@@ -11297,19 +11327,33 @@ opt_limit_clause:
         | limit_clause {}
         ;
 
+limit_clause_init:
+          LIMIT
+          {
+            SELECT_LEX *sel= Select;
+            if (sel->master_unit()->is_union() && !sel->braces)
+            {
+              /* Move LIMIT that belongs to UNION to fake_select_lex */
+              Lex->current_select= sel->master_unit()->fake_select_lex;
+              DBUG_ASSERT(Select);
+            }
+          }
+        ;  
+
 limit_clause:
-          LIMIT limit_options
+          limit_clause_init limit_options
           {
             SELECT_LEX *sel= Select;
             if (!sel->select_limit->basic_const_item() ||
                 sel->select_limit->val_int() > 0)
               Lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_LIMIT);
           }
-        | LIMIT limit_options ROWS_SYM EXAMINED_SYM limit_rows_option
+        | limit_clause_init limit_options
+          ROWS_SYM EXAMINED_SYM limit_rows_option
           {
             Lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_LIMIT);
           }
-        | LIMIT ROWS_SYM EXAMINED_SYM limit_rows_option
+        | limit_clause_init ROWS_SYM EXAMINED_SYM limit_rows_option
           {
             Lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_LIMIT);
           }
@@ -12790,6 +12834,7 @@ flush_options:
             YYPS->m_mdl_type= MDL_SHARED_HIGH_PRIO;
           }
           opt_table_list opt_flush_lock
+          {}
         | flush_options_list
         ;
 
@@ -15739,7 +15784,6 @@ union_order_or_limit:
             SELECT_LEX *fake= unit->fake_select_lex;
             if (fake)
             {
-              unit->global_parameters= fake;
               fake->no_table_names_allowed= 1;
               lex->current_select= fake;
             }