Fixed the problem of wrong identification of WITH tables defined in WITH clauses without RECURSIVE.

Added test cases to check the fix.
Fixed the problem of wrong types of recursive tables when the type of anchor part does not coincide with the
type of recursive part.
Prevented usage of marerialization and subquery cache for subqueries with recursive references.
Introduced system variables 'max_recursion_level'.
Added a test case to test usage of this variable.

11 files changed, 219 insertions, 132 deletions

diff --git a/sql/item_subselect.cc b/sql/item_subselect.cc
index 94e7bc98618..7d458282825 100644
--- a/sql/item_subselect.cc
+++ b/sql/item_subselect.cc
@@ -54,7 +54,7 @@ Item_subselect::Item_subselect(THD *thd_arg):
   have_to_be_excluded(0),
   inside_first_fix_fields(0), done_first_fix_fields(FALSE), 
   expr_cache(0), forced_const(FALSE), substitution(0), engine(0), eliminated(FALSE),
-  changed(0), is_correlated(FALSE)
+  changed(0), is_correlated(FALSE), with_recursive_reference(0)
 {
   DBUG_ENTER("Item_subselect::Item_subselect");
   DBUG_PRINT("enter", ("this: 0x%lx", (ulong) this));
@@ -771,7 +771,8 @@ bool Item_subselect::expr_cache_is_needed(THD *thd)
           engine->cols() == 1 &&
           optimizer_flag(thd, OPTIMIZER_SWITCH_SUBQUERY_CACHE) &&
           !(engine->uncacheable() & (UNCACHEABLE_RAND |
-                                     UNCACHEABLE_SIDEEFFECT)));
+                                     UNCACHEABLE_SIDEEFFECT)) &&
+          !with_recursive_reference);
 }
 
 
@@ -810,7 +811,8 @@ bool Item_in_subselect::expr_cache_is_needed(THD *thd)
 {
   return (optimizer_flag(thd, OPTIMIZER_SWITCH_SUBQUERY_CACHE) &&
           !(engine->uncacheable() & (UNCACHEABLE_RAND |
-                                     UNCACHEABLE_SIDEEFFECT)));
+                                     UNCACHEABLE_SIDEEFFECT)) &&
+          !with_recursive_reference);
 }
 
 
diff --git a/sql/item_subselect.h b/sql/item_subselect.h
index 58b5a948048..c1e68247220 100644
--- a/sql/item_subselect.h
+++ b/sql/item_subselect.h
@@ -126,7 +126,9 @@ public:
   bool changed;
 
   /* TRUE <=> The underlying SELECT is correlated w.r.t some ancestor select */
-  bool is_correlated; 
+  bool is_correlated;
+
+  bool with_recursive_reference; 
 
   enum subs_type {UNKNOWN_SUBS, SINGLEROW_SUBS,
 		  EXISTS_SUBS, IN_SUBS, ALL_SUBS, ANY_SUBS};
diff --git a/sql/opt_subselect.cc b/sql/opt_subselect.cc
index 55c6c075f48..afb439040de 100644
--- a/sql/opt_subselect.cc
+++ b/sql/opt_subselect.cc
@@ -512,6 +512,7 @@ bool is_materialization_applicable(THD *thd, Item_in_subselect *in_subs,
         (Subquery is correlated to the immediate outer query &&
          Subquery !contains {GROUP BY, ORDER BY [LIMIT],
          aggregate functions}) && subquery predicate is not under "NOT IN"))
+    5. Subquery does not contain recursive references
 
   A note about prepared statements: we want the if-branch to be taken on
   PREPARE and each EXECUTE. The rewrites are only done once, but we need 
@@ -528,7 +529,8 @@ bool is_materialization_applicable(THD *thd, Item_in_subselect *in_subs,
                         OPTIMIZER_SWITCH_PARTIAL_MATCH_ROWID_MERGE) || //3
          optimizer_flag(thd,
                         OPTIMIZER_SWITCH_PARTIAL_MATCH_TABLE_SCAN)) && //3
-        !in_subs->is_correlated)                                       //4
+        !in_subs->is_correlated &&                                     //4
+        !in_subs->with_recursive_reference)                            //5
    {
      return TRUE;
    }
diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index 00228ee1062..4bacee0d9f3 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/sql/share/errmsg-utf8.txt
@@ -7150,8 +7150,6 @@ ER_WITH_COL_WRONG_LIST
         eng "WITH column list and SELECT field list have different column counts"
 ER_DUP_QUERY_NAME
         eng "Duplicate query name in WITH clause"
-ER_WRONG_ORDER_IN_WITH_CLAUSE
-        eng "The definition of the table '%s' refers to the table '%s' defined later in a non-recursive WITH clause"
 ER_RECURSIVE_WITHOUT_ANCHORS
         eng "No anchors for recursive WITH element '%s'"
 ER_REF_TO_RECURSIVE_WITH_TABLE_IN_DERIVED
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 7e995c04b33..04ca37295bb 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -558,6 +558,7 @@ typedef struct system_variables
   ulong max_allowed_packet;
   ulong max_error_count;
   ulong max_length_for_sort_data;
+  ulong max_recursion_level;
   ulong max_sort_length;
   ulong max_tmp_tables;
   ulong max_insert_delayed_threads;
diff --git a/sql/sql_cte.cc b/sql/sql_cte.cc
index 77d2c7d24d3..04a4fcb8a2b 100644
--- a/sql/sql_cte.cc
+++ b/sql/sql_cte.cc
@@ -118,83 +118,78 @@ bool With_clause::check_dependencies(THD *thd)
     if (with_elem->derived_dep_map & with_elem->get_elem_map())
       with_elem->is_recursive= true;
   }   
-  for (With_element *with_elem= first_elem;
-       with_elem != NULL;
-       with_elem= with_elem->next_elem)
-  {
-    if (with_elem->is_recursive)
-    {
-#if 0  
-      my_error(ER_RECURSIVE_QUERY_IN_WITH_CLAUSE, MYF(0),
-               with_elem->query_name->str);
-      return true;
-#endif
-    }
-  }
-
-  if (!with_recursive)
-  {
-    /* 
-      For each with table T defined in this with clause check whether
-      it is used in any definition that follows the definition of T.
-    */
-    for (With_element *with_elem= first_elem;
-         with_elem != NULL;
-         with_elem= with_elem->next_elem)
-    {
-      With_element *checked_elem= with_elem->next_elem;
-      for (uint i = with_elem->number+1;
-           i < elements;
-           i++, checked_elem= checked_elem->next_elem)
-      {
-        if (with_elem->check_dependency_on(checked_elem))
-        {
-          my_error(ER_WRONG_ORDER_IN_WITH_CLAUSE, MYF(0),
-                   with_elem->query_name->str, checked_elem->query_name->str);
-          return true;
-        }
-      }
-    }
-  }
 	
   dependencies_are_checked= true;
   return false;
 }
 
 
+struct st_unit_ctxt_elem
+{
+  st_unit_ctxt_elem *prev;
+  st_select_lex_unit *unit;
+};
+
 bool With_element::check_dependencies_in_spec(THD *thd)
 { 
   for (st_select_lex *sl=  spec->first_select(); sl; sl= sl->next_select())
   {
-    check_dependencies_in_select(sl, sl->with_dep);
+    st_unit_ctxt_elem ctxt0= {NULL, owner->owner};
+    st_unit_ctxt_elem ctxt1= {&ctxt0, spec};
+    check_dependencies_in_select(sl, &ctxt1, false, &sl->with_dep);
     base_dep_map|= sl->with_dep;
   }
   return false;
 }
 
 
+With_element *find_table_def_in_with_clauses(TABLE_LIST *tbl,
+                                             st_unit_ctxt_elem *ctxt)
+{
+  With_element *barrier= NULL;
+  for (st_unit_ctxt_elem *unit_ctxt_elem= ctxt;
+       unit_ctxt_elem;
+       unit_ctxt_elem= unit_ctxt_elem->prev)
+  {
+    st_select_lex_unit *unit= unit_ctxt_elem->unit;
+    With_clause *with_clause= unit->with_clause;
+    if (with_clause &&
+	(tbl->with= with_clause->find_table_def(tbl, barrier)))
+      return tbl->with;
+    barrier= NULL;
+    if (unit->with_element && !unit->with_element->get_owner()->with_recursive)
+      barrier= unit->with_element;
+  }
+  return NULL;
+}
+
+
 void With_element::check_dependencies_in_select(st_select_lex *sl,
-                                                table_map &dep_map)
+                                                st_unit_ctxt_elem *ctxt,
+                                                bool in_subq,
+                                                table_map *dep_map)
 {
-  bool is_sq_select= sl->master_unit()->item != NULL;
+  With_clause *with_clause= sl->get_with_clause();
   for (TABLE_LIST *tbl= sl->table_list.first; tbl; tbl= tbl->next_local)
   {
+    if (tbl->derived || tbl->nested_join)
+      continue;
     tbl->with_internal_reference_map= 0;
+    if (with_clause && !tbl->with)
+      tbl->with= with_clause->find_table_def(tbl, NULL);
     if (!tbl->with)
-      tbl->with= owner->find_table_def(tbl);
-    if (!tbl->with && tbl->select_lex)
-      tbl->with= tbl->select_lex->find_table_def_in_with_clauses(tbl);
+      tbl->with= find_table_def_in_with_clauses(tbl, ctxt);
     if (tbl->with && tbl->with->owner== this->owner)
     {
-      dep_map|= tbl->with->get_elem_map();
+      *dep_map|= tbl->with->get_elem_map();
       tbl->with_internal_reference_map= get_elem_map();
-      if (is_sq_select)
+      if (in_subq)
         sq_dep_map|= tbl->with->get_elem_map();
     }
   }
   st_select_lex_unit *inner_unit= sl->first_inner_unit();
   for (; inner_unit; inner_unit= inner_unit->next_unit())
-    check_dependencies_in_unit(inner_unit, dep_map);
+    check_dependencies_in_unit(inner_unit, ctxt, in_subq, dep_map);
 }
 
 
@@ -213,12 +208,32 @@ void With_element::check_dependencies_in_select(st_select_lex *sl,
 */
 
 void With_element::check_dependencies_in_unit(st_select_lex_unit *unit,
-                                              table_map &dep_map)
+                                              st_unit_ctxt_elem *ctxt,
+                                              bool in_subq,
+                                              table_map *dep_map)
 {
+  if (unit->with_clause)
+    check_dependencies_in_with_clause(unit->with_clause, ctxt, in_subq, dep_map);
+  in_subq |= unit->item != NULL;
+  st_unit_ctxt_elem unit_ctxt_elem= {ctxt, unit};
   st_select_lex *sl= unit->first_select();
   for (; sl; sl= sl->next_select())
   {
-    check_dependencies_in_select(sl, dep_map);
+    check_dependencies_in_select(sl, &unit_ctxt_elem, in_subq, dep_map);
+  }
+}
+
+void 
+With_element::check_dependencies_in_with_clause(With_clause *with_clause,
+                                                st_unit_ctxt_elem *ctxt,
+                                                bool in_subq,
+                                                table_map *dep_map)
+{
+  for (With_element *with_elem= with_clause->first_elem;
+       with_elem != NULL;
+       with_elem= with_elem->next_elem)
+  {
+    check_dependencies_in_unit(with_elem->spec, ctxt, in_subq, dep_map);
   }
 }
 
@@ -328,10 +343,11 @@ bool With_clause::check_anchors()
     NULL - otherwise
 */    
 
-With_element *With_clause::find_table_def(TABLE_LIST *table)
+With_element *With_clause::find_table_def(TABLE_LIST *table,
+                                          With_element *barrier)
 {
   for (With_element *with_elem= first_elem; 
-       with_elem != NULL;
+       with_elem != barrier;
        with_elem= with_elem->next_elem)
   {
     if (my_strcasecmp(system_charset_info, with_elem->query_name->str,
@@ -672,17 +688,27 @@ bool With_element::is_anchor(st_select_lex *sel)
 
 With_element *st_select_lex::find_table_def_in_with_clauses(TABLE_LIST *table)
 {
+  st_select_lex_unit *master_unit= NULL;
   With_element *found= NULL;
   for (st_select_lex *sl= this;
        sl;
-       sl= sl->master_unit()->outer_select())
+       sl= master_unit->outer_select())
   {
+    With_element *with_elem= sl->get_with_element();
+    /* 
+      If sl->master_unit() is the spec of a with element then the search for 
+      a definition was already done by With_element::check_dependencies_in_spec
+      and it was unsuccesful.
+    */
+    if (with_elem)
+      break;      
     With_clause *with_clause=sl->get_with_clause();
-    if (with_clause && (found= with_clause->find_table_def(table)))
-      return found; 
+    if (with_clause && (found= with_clause->find_table_def(table,NULL)))
+      break;
+    master_unit= sl->master_unit();
     /* Do not look for the table's definition beyond the scope of the view */
-    if (sl->master_unit()->is_view)
-      break;      
+    if (master_unit->is_view)
+      break; 
   }
   return found;
 }
@@ -729,7 +755,7 @@ bool TABLE_LIST::is_recursive_with_table()
 bool TABLE_LIST::is_with_table_recursive_reference()
 {
   return (with_internal_reference_map &&
-          (with->mutually_recursive & with_internal_reference_map));
+          (with->get_mutually_recursive() & with_internal_reference_map));
 }
 
 
@@ -745,10 +771,11 @@ bool st_select_lex::check_unrestricted_recursive(bool only_standards_compliant)
 					      unrestricted, 
 					      encountered))
     return true;
-  with_elem->owner->unrestricted|= unrestricted;
+  with_elem->get_owner()->add_unrestricted(unrestricted);
   if (with_sum_func ||
-      (with_elem->sq_dep_map & with_elem->mutually_recursive))
-    with_elem->owner->unrestricted|= with_elem->mutually_recursive;
+      (with_elem->contains_sq_with_recursive_reference()))
+    with_elem->get_owner()->add_unrestricted(
+                              with_elem->get_mutually_recursive());
   if (only_standards_compliant && with_elem->is_unrestricted())
   {
     my_error(ER_NOT_STANDARDS_COMPLIANT_RECURSIVE,
@@ -776,7 +803,7 @@ bool With_element::check_unrestricted_recursive(st_select_lex *sel,
         if (tbl->is_materialized_derived())
         {
           table_map dep_map;
-          check_dependencies_in_unit(unit, dep_map);
+          check_dependencies_in_unit(unit, NULL, false, &dep_map);
           if (dep_map & get_elem_map())
           {
 	    my_error(ER_REF_TO_RECURSIVE_WITH_TABLE_IN_DERIVED,
@@ -797,7 +824,7 @@ bool With_element::check_unrestricted_recursive(st_select_lex *sel,
       else
         encountered|= with_elem->get_elem_map();
     }
-  }
+  } 
   for (With_element *with_elem= sel->get_with_element()->owner->first_elem;
        with_elem != NULL;
        with_elem= with_elem->next_elem)
@@ -841,6 +868,27 @@ bool With_element::check_unrestricted_recursive(st_select_lex *sel,
 }
 
 
+void st_select_lex::check_subqueries_with_recursive_references()
+{
+  st_select_lex_unit *sl_master= master_unit();
+  List_iterator<TABLE_LIST> ti(leaf_tables);
+  TABLE_LIST *tbl;
+  while ((tbl= ti++))
+  {
+    if (!(tbl->is_with_table_recursive_reference() && sl_master->item))
+      continue;
+    for (st_select_lex *sl= this; sl; sl= sl_master->outer_select())
+    { 
+      sl_master= sl->master_unit();
+      if (!sl_master->item)
+	continue;
+      Item_subselect *subq= (Item_subselect *) sl_master->item;
+      subq->with_recursive_reference= true;
+    }
+  }
+}
+
+
 /**
   @brief
     Print this with clause
diff --git a/sql/sql_cte.h b/sql/sql_cte.h
index 1c32f16258c..23eea8463e6 100644
--- a/sql/sql_cte.h
+++ b/sql/sql_cte.h
@@ -3,8 +3,8 @@
 #include "sql_list.h"
 #include "sql_lex.h"
 
-class With_clause;
 class select_union;
+struct st_unit_ctxt_elem;
 
 /**
   @class With_clause
@@ -100,10 +100,19 @@ public:
 
   bool check_dependencies_in_spec(THD *thd);
   
-  void check_dependencies_in_select(st_select_lex *sl, table_map &dep_map);
+  void check_dependencies_in_select(st_select_lex *sl, st_unit_ctxt_elem *ctxt,
+                                    bool in_subq, table_map *dep_map);
       
-  void check_dependencies_in_unit(st_select_lex_unit *unit, table_map &dep_map);
- 
+  void check_dependencies_in_unit(st_select_lex_unit *unit,
+                                  st_unit_ctxt_elem *ctxt,
+                                  bool in_subq,
+                                  table_map *dep_map);
+
+  void check_dependencies_in_with_clause(With_clause *with_clause, 
+                                         st_unit_ctxt_elem *ctxt,
+                                         bool in_subq,
+                                         table_map *dep_map);
+
   void  set_dependency_on(With_element *with_elem)
   { base_dep_map|= with_elem->get_elem_map(); }
 
@@ -126,7 +135,14 @@ public:
                                     table_map &unrestricted,
                                     table_map &encountered);
 
-   void print(String *str, enum_query_type query_type);
+  void print(String *str, enum_query_type query_type);
+
+  With_clause *get_owner() { return owner; }
+
+  bool contains_sq_with_recursive_reference()
+  { return sq_dep_map & mutually_recursive; }
+
+  table_map get_mutually_recursive() { return mutually_recursive; }
 
   void set_table(TABLE *tab) { table= tab; }
 
@@ -151,11 +167,6 @@ public:
   void set_result_table(TABLE *tab) { result_table= tab; }
 
   friend class With_clause;
-  friend
-  bool 
-  st_select_lex::check_unrestricted_recursive(bool only_standard_compliant);
-  friend
-  bool TABLE_LIST::is_with_table_recursive_reference();
 };
 
 
@@ -209,8 +220,7 @@ public:
   { 
     elem->owner= this;
     elem->number= elements;
-    owner= elem->spec;
-    owner->with_element= elem;
+    elem->spec->with_element= elem;
     *last_next= elem;
     last_next= &elem->next_elem;
     elements++;
@@ -224,6 +234,8 @@ public:
     last_next= &this->next_with_clause;
   }
 
+  void set_owner(st_select_lex_unit *unit) { owner= unit; }
+
   With_clause *pop() { return embedding_with_clause; }
       
   bool check_dependencies(THD *thd);
@@ -232,12 +244,14 @@ public:
 
   void move_anchors_ahead();
 
-  With_element *find_table_def(TABLE_LIST *table);
+  With_element *find_table_def(TABLE_LIST *table, With_element *barrier);
 
   With_element *find_table_def_in_with_clauses(TABLE_LIST *table);
 
   bool prepare_unreferenced_elements(THD *thd);
 
+  void add_unrestricted(table_map map) { unrestricted|= map; }
+
   void print(String *str, enum_query_type query_type);
 
   friend class With_element;
@@ -245,10 +259,6 @@ public:
   friend
   bool
   check_dependencies_in_with_clauses(THD *thd, With_clause *with_clauses_list);
-  friend
-  bool
-  st_select_lex::check_unrestricted_recursive(bool only_standard_compliant);
-
 };
 
 inline
@@ -292,5 +302,20 @@ void With_element::reset_for_exec()
   owner->cleaned&= ~get_elem_map();    
 }
 
+inline
+void  st_select_lex_unit::set_with_clause(With_clause *with_cl)
+{ 
+    with_clause= with_cl;
+    if (with_clause)
+      with_clause->set_owner(this);
+}
+
+inline
+void st_select_lex::set_with_clause(With_clause *with_clause)
+{
+  master_unit()->with_clause= with_clause;
+  if (with_clause)
+    with_clause->set_owner(master_unit());
+}
 
 #endif /* SQL_CTE_INCLUDED */
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index b17e19276da..762d6718dcb 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -30,6 +30,7 @@
 #include "sql_alter.h"                // Alter_info
 #include "sql_window.h"
 
+
 /* YACC and LEX Definitions */
 
 /* These may not be declared yet */
@@ -690,7 +691,7 @@ public:
   {
     return reinterpret_cast<st_select_lex*>(slave);
   }
-  void set_with_clause(With_clause *with_cl) { with_clause= with_cl; }
+  void set_with_clause(With_clause *with_cl);
   st_select_lex_unit* next_unit()
   {
     return reinterpret_cast<st_select_lex_unit*>(next);
@@ -1095,10 +1096,7 @@ public:
 
   void set_non_agg_field_used(bool val) { m_non_agg_field_used= val; }
   void set_agg_func_used(bool val)      { m_agg_func_used= val; }
-  void set_with_clause(With_clause *with_clause)
-  {
-    master_unit()->with_clause= with_clause;
-  }
+  void set_with_clause(With_clause *with_clause);
   With_clause *get_with_clause()
   {
     return master_unit()->with_clause;
@@ -1109,8 +1107,8 @@ public:
   }
   With_element *find_table_def_in_with_clauses(TABLE_LIST *table);
   bool check_unrestricted_recursive(bool only_standards_compliant); 
-
-
+  void check_subqueries_with_recursive_references();
+  
   List<Window_spec> window_specs;
   void prepare_add_window_spec(THD *thd);
   bool add_window_def(THD *thd, LEX_STRING *win_name, LEX_STRING *win_ref,
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 6792d7f5e2c..84dd2e4b676 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -865,6 +865,7 @@ JOIN::prepare(TABLE_LIST *tables_init,
       select_lex->check_unrestricted_recursive(
                       thd->variables.only_standards_compliant_cte))
     DBUG_RETURN(-1);
+  select_lex->check_subqueries_with_recursive_references();
   
   int res= check_and_do_in_subquery_rewrites(this);
 
diff --git a/sql/sql_union.cc b/sql/sql_union.cc
index c43fdf30a64..d19bbaf103c 100644
--- a/sql/sql_union.cc
+++ b/sql/sql_union.cc
@@ -442,6 +442,7 @@ bool st_select_lex_unit::prepare(THD *thd_arg, select_result *sel_result,
   SELECT_LEX *lex_select_save= thd_arg->lex->current_select;
   SELECT_LEX *sl, *first_sl= first_select();
   bool is_recursive= with_element && with_element->is_recursive;
+  bool is_rec_result_table_created= false;
   select_result *tmp_result;
   bool is_union_select;
   bool instantiate_tmp_table= false;
@@ -609,24 +610,6 @@ bool st_select_lex_unit::prepare(THD *thd_arg, select_result *sel_result,
 
       if (thd_arg->is_fatal_error)
 	goto err; // out of memory
-
-      if (is_recursive)
-      {
-
-        ulonglong create_options;
-        create_options= (first_sl->options | thd_arg->variables.option_bits |
-                     TMP_TABLE_ALL_COLUMNS);
-        if (union_result->create_result_table(thd, &types,
-                                              MY_TEST(union_distinct),
-                                              create_options, derived->alias,
-                                              false,
-                                              instantiate_tmp_table, false))
-          goto err;
-        if (!derived->table)
-          derived->table= derived->derived_result->table= 
-            with_element->rec_result->rec_tables.head();
-        with_element->mark_as_with_prepared_anchor();      
-      }
     }
     else
     {
@@ -636,19 +619,42 @@ bool st_select_lex_unit::prepare(THD *thd_arg, select_result *sel_result,
 		   ER_THD(thd, ER_WRONG_NUMBER_OF_COLUMNS_IN_SELECT),MYF(0));
 	goto err;
       }
-      List_iterator_fast<Item> it(sl->item_list);
-      List_iterator_fast<Item> tp(types);	
-      Item *type, *item_tmp;
-      while ((type= tp++, item_tmp= it++))
+      if (!is_rec_result_table_created)
       {
-        if (((Item_type_holder*)type)->join_types(thd_arg, item_tmp))
-	  DBUG_RETURN(TRUE);
+        List_iterator_fast<Item> it(sl->item_list);
+        List_iterator_fast<Item> tp(types);	
+        Item *type, *item_tmp;
+        while ((type= tp++, item_tmp= it++))
+        {
+          if (((Item_type_holder*)type)->join_types(thd_arg, item_tmp))
+	    DBUG_RETURN(TRUE);
+        }
       }
     }
-    if (with_element && !with_element->is_anchor(sl))
+    if (is_recursive)
     {
-      sl->uncacheable|= UNCACHEABLE_UNITED;
-    } 
+      if (!with_element->is_anchor(sl))
+        sl->uncacheable|= UNCACHEABLE_UNITED;
+      if(!is_rec_result_table_created &&
+         (!sl->next_select() ||
+          sl->next_select() == with_element->first_recursive))
+      {
+        ulonglong create_options;
+        create_options= (first_sl->options | thd_arg->variables.option_bits |
+                         TMP_TABLE_ALL_COLUMNS);
+        if (union_result->create_result_table(thd, &types,
+                                              MY_TEST(union_distinct),
+                                              create_options, derived->alias,
+                                              false,
+                                              instantiate_tmp_table, false))
+          goto err;
+        if (!derived->table)
+          derived->table= derived->derived_result->table= 
+            with_element->rec_result->rec_tables.head();
+        with_element->mark_as_with_prepared_anchor();
+        is_rec_result_table_created= true;
+      }
+    }      
   }
 
   /*
@@ -1166,17 +1172,18 @@ bool st_select_lex_unit::exec_recursive()
   st_select_lex *first_recursive_sel= with_element->first_recursive;
   TABLE *incr_table= with_element->rec_result->incr_table;
   TABLE *result_table= with_element->result_table;
-  ha_rows last_union_records= 0;
   ha_rows examined_rows= 0;
   bool unrestricted= with_element->is_unrestricted();
-  bool is_stabilized= false;
-  DBUG_ENTER("st_select_lex_unit::exec_recursive");
+  bool no_more_iterations= false;
   bool with_anchor= with_element->with_anchor;
   st_select_lex *first_sl= first_select();
   st_select_lex *barrier= with_anchor ? first_recursive_sel : NULL;
+  uint max_level= thd->variables.max_recursion_level;
   List_iterator_fast<TABLE> li(with_element->rec_result->rec_tables);
   TABLE *rec_table;
 
+  DBUG_ENTER("st_select_lex_unit::exec_recursive");
+
   do
   {
     if ((saved_error= incr_table->file->ha_delete_all_rows()))
@@ -1210,16 +1217,13 @@ bool st_select_lex_unit::exec_recursive()
       barrier= NULL;
     }
 
-    table->file->info(HA_STATUS_VARIABLE);
-    if (table->file->stats.records ==  last_union_records)
-    {
-       is_stabilized= true;
-    }
+    incr_table->file->info(HA_STATUS_VARIABLE);
+    if (incr_table->file->stats.records == 0 ||
+        with_element->level + 1 == max_level)
+       no_more_iterations= true;
     else
-    {
-      last_union_records= table->file->stats.records;
       with_element->level++;
-    }
+
     li.rewind();
     while ((rec_table= li++))
     {
@@ -1227,7 +1231,7 @@ bool st_select_lex_unit::exec_recursive()
                                                          !unrestricted)))
 	  goto err;
     }
-  } while (!is_stabilized); 
+  } while (!no_more_iterations); 
 
   if ((saved_error= table->insert_all_rows_into(thd,
                                                 result_table,
diff --git a/sql/sys_vars.cc b/sql/sys_vars.cc
index c921fffc004..f63549ba3d7 100644
--- a/sql/sys_vars.cc
+++ b/sql/sys_vars.cc
@@ -2145,6 +2145,12 @@ static Sys_var_ulong Sys_max_prepared_stmt_count(
        VALID_RANGE(0, 1024*1024), DEFAULT(16382), BLOCK_SIZE(1),
        &PLock_prepared_stmt_count);
 
+static Sys_var_ulong Sys_max_recursion_level(
+       "max_recursion_level",
+       "Maximum number of iterations when executing recursive queries",
+       SESSION_VAR(max_recursion_level), CMD_LINE(OPT_ARG),
+       VALID_RANGE(0, UINT_MAX), DEFAULT(UINT_MAX), BLOCK_SIZE(1));
+
 static Sys_var_ulong Sys_max_sort_length(
        "max_sort_length",
        "The number of bytes to use when sorting BLOB or TEXT values (only "