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.

mysql-test/r/cte_nonrecursive.result

@@ -675,7 +675,7 @@ ERROR HY000: Duplicate query name in WITH clause
 with t as (select a from s where a<5),
 s as (select a from t1 where b>='d')
 select * from t,s where t.a=s.a;
-ERROR HY000: The definition of the table 't' refers to the table 's' defined later in a non-recursive WITH clause
+ERROR 42S02: Table 'test.s' doesn't exist
 with recursive
 t as (select a from s where a<5),
 s as (select a from t1 where b>='d')
@@ -709,7 +709,8 @@ with recursive
 t as (select * from t1 where b>'aaa' and b <='d') 
 select t.b from t,t2
 where t.a=t2.c and
-t2.c in (with s as (select t1.a from s,t1 where t1.a=s.a and t1.b<'c')
+t2.c in (with recursive
+s as (select t1.a from s,t1 where t1.a=s.a and t1.b<'c')
 select * from s);
 ERROR HY000: No anchors for recursive WITH element 's'
 #erroneous definition of unreferenced with table t

mysql-test/r/cte_recursive.result

@@ -21,6 +21,116 @@ select * from b1 where b1.b > 'auu')
 select * from c1;
 ERROR HY000: No anchors for recursive WITH element 'b1'
 drop table t1;
+# WITH RECURSIVE vs just WITH
+create table t1 (a int);
+insert into t1 values
+(0), (1), (2), (3), (4);
+create table t2 (a int);
+insert into t2 values
+(1), (2), (3), (4), (5);
+# just WITH : s refers to t defined after s
+with 
+s(a) as (select t.a + 10 from t),
+t(a) as (select t1.a from t1)
+select * from s;
+ERROR 42S02: Table 'test.t' doesn't exist
+# WITH RECURSIVE: s refers to t defined after s
+with recursive 
+s(a) as (select t.a + 10 from t),
+t(a) as (select t1.a from t1)
+select * from s;
+a
+10
+11
+12
+13
+14
+# just WITH : defined t1 is non-recursive and uses base tables t1,t2
+with
+t1 as
+(
+select a  from t2 where t2.a=3
+union
+select t2.a from t1,t2 where t1.a+1=t2.a
+)
+select * from t1;
+a
+3
+1
+2
+4
+5
+explain
+with
+t1 as
+(
+select a  from t2 where t2.a=3
+union
+select t2.a from t1,t2 where t1.a+1=t2.a
+)
+select * from t1;
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	PRIMARY	<derived2>	ALL	NULL	NULL	NULL	NULL	30	
+2	SUBQUERY	t2	ALL	NULL	NULL	NULL	NULL	5	Using where
+3	UNION	t1	ALL	NULL	NULL	NULL	NULL	5	
+3	UNION	t2	ALL	NULL	NULL	NULL	NULL	5	Using where; Using join buffer (flat, BNL join)
+NULL	UNION RESULT	<union2,3>	ALL	NULL	NULL	NULL	NULL	NULL	
+#WITH RECURSIVE : defined t1 is recursive and uses only base table t2
+with recursive
+t1 as
+(
+select a  from t2 where t2.a=3
+union
+select t2.a from t1,t2 where t1.a+1=t2.a
+)
+select * from t1;
+a
+3
+4
+5
+explain
+with recursive
+t1 as
+(
+select a  from t2 where t2.a=3
+union
+select t2.a from t1,t2 where t1.a+1=t2.a
+)
+select * from t1;
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	PRIMARY	<derived2>	ALL	NULL	NULL	NULL	NULL	30	
+2	SUBQUERY	t2	ALL	NULL	NULL	NULL	NULL	5	Using where
+3	UNCACHEABLE UNION	<derived2>	ALL	NULL	NULL	NULL	NULL	5	
+3	UNCACHEABLE UNION	t2	ALL	NULL	NULL	NULL	NULL	5	Using where; Using join buffer (flat, BNL join)
+NULL	UNION RESULT	<union2,3>	ALL	NULL	NULL	NULL	NULL	NULL	
+# just WITH : types of t1 columns are determined by all parts of union 
+create view v1 as
+with
+t1 as
+(
+select a  from t2 where t2.a=3
+union
+select t2.a+1 from t1,t2 where t1.a=t2.a
+)
+select * from t1;
+show columns from v1;
+Field	Type	Null	Key	Default	Extra
+a	bigint(20)	YES		NULL	
+# WITH RECURSIVE : types of t1 columns are determined by anchor parts 
+create view v2 as
+with recursive
+t1 as
+(
+select a  from t2 where t2.a=3
+union
+select t2.a+1 from t1,t2 where t1.a=t2.a
+)
+select * from t1;
+show columns from v2;
+Field	Type	Null	Key	Default	Extra
+a	int(11)	YES		NULL	
+drop view v1,v2;
+drop table t1,t2;
 create table  folks(id int, name char(32), dob date, father int, mother int);
 insert into folks values
 (100, 'Me', '2000-01-01', 20, 30),
@@ -485,7 +595,6 @@ select generation, name from ancestor_ids a, folks
 where a.id = folks.id;
 ERROR HY000: Restrictions imposed on recursive definitions are violated for table 'ancestor_ids'
 set standards_compliant_cte=0;
-set optimizer_switch='materialization=off,subquery_cache=off';
 with recursive
 ancestor_ids (id, generation)
 as
@@ -529,7 +638,6 @@ generation	name
 2	Grandma Ann
 2	Grandma Sally
 3	Grandgrandma Martha
-set optimizer_switch=default;
 set standards_compliant_cte=1;
 with recursive
 coupled_ancestor_ids (id)
@@ -716,4 +824,33 @@ generation	name
 1	Mom
 2	Grandpa Bill
 2	Grandma Ann
+set statement max_recursion_level=2 for
+with recursive
+ancestor_ids (id, generation)
+as
+(
+select father, 1 from folks where name = 'Me'
+  union
+select mother, 1 from folks where name = 'Me'
+  union
+select father, a.generation+1 from folks, ancestor_ids a
+where folks.id = a.id
+union
+select mother, a.generation+1 from folks, ancestor_ids a
+where folks.id = a.id
+),
+ancestors 
+as
+(
+select generation, name from folks as p, ancestor_ids as a
+where p.id = a.id
+)
+select * from ancestors;
+generation	name
+1	Dad
+1	Mom
+2	Grandpa Bill
+2	Grandpa Ben
+2	Grandma Ann
+2	Grandma Sally
 drop table folks;

mysql-test/r/mysqld--help.result

@@ -450,6 +450,9 @@ The following options may be given as the first argument:
  max_allowed_packet instead.
  --max-prepared-stmt-count=# 
  Maximum number of prepared statements in the server
+ --max-recursion-level[=#] 
+ Maximum number of iterations when executing recursive
+ queries
  --max-relay-log-size=# 
  relay log will be rotated automatically when the size
  exceeds this value.  If 0 at startup, it's set to
@@ -1270,6 +1273,7 @@ max-join-size 18446744073709551615
 max-length-for-sort-data 1024
 max-long-data-size 4194304
 max-prepared-stmt-count 16382
+max-recursion-level 18446744073709551615
 max-relay-log-size 1073741824
 max-seeks-for-key 18446744073709551615
 max-sort-length 1024

mysql-test/t/cte_nonrecursive.test

@@ -366,7 +366,7 @@ with t as (select * from t2 where c>3),
      t as (select a from t1 where a>2)
  select * from t,t1 where t1.a=t.c;
 
---ERROR ER_WRONG_ORDER_IN_WITH_CLAUSE
+--ERROR ER_NO_SUCH_TABLE
 with t as (select a from s where a<5),
      s as (select a from t1 where b>='d')
   select * from t,s where t.a=s.a;
@@ -402,8 +402,9 @@ with recursive
   t as (select * from t1 where b>'aaa' and b <='d') 
   select t.b from t,t2
     where t.a=t2.c and
-          t2.c in (with s as (select t1.a from s,t1 where t1.a=s.a and t1.b<'c')
-                     select * from s);
+          t2.c in (with recursive
+                   s as (select t1.a from s,t1 where t1.a=s.a and t1.b<'c')
+                   select * from s);
 --echo #erroneous definition of unreferenced with table t
 --ERROR ER_BAD_FIELD_ERROR
 with t as (select count(*) from t1 where d>='f' group by a)

mysql-test/t/cte_recursive.test

@@ -24,6 +24,105 @@ select * from c1;
 
 drop table t1;
 
+
+--echo # WITH RECURSIVE vs just WITH
+
+create table t1 (a int);
+insert into t1 values
+  (0), (1), (2), (3), (4);
+create table t2 (a int);
+insert into t2 values
+  (1), (2), (3), (4), (5);
+
+
+--echo # just WITH : s refers to t defined after s
+--ERROR ER_NO_SUCH_TABLE
+with 
+ s(a) as (select t.a + 10 from t),
+ t(a) as (select t1.a from t1)
+select * from s;
+
+--echo # WITH RECURSIVE: s refers to t defined after s
+with recursive 
+ s(a) as (select t.a + 10 from t),
+ t(a) as (select t1.a from t1)
+select * from s;
+
+--echo # just WITH : defined t1 is non-recursive and uses base tables t1,t2
+with
+t1 as
+(
+select a  from t2 where t2.a=3
+union
+select t2.a from t1,t2 where t1.a+1=t2.a
+)
+select * from t1;
+
+explain
+with
+t1 as
+(
+select a  from t2 where t2.a=3
+union
+select t2.a from t1,t2 where t1.a+1=t2.a
+)
+select * from t1;
+
+
+--echo #WITH RECURSIVE : defined t1 is recursive and uses only base table t2
+with recursive
+t1 as
+(
+select a  from t2 where t2.a=3
+union
+select t2.a from t1,t2 where t1.a+1=t2.a
+)
+select * from t1;
+
+explain
+with recursive
+t1 as
+(
+select a  from t2 where t2.a=3
+union
+select t2.a from t1,t2 where t1.a+1=t2.a
+)
+select * from t1;
+    
+--echo # just WITH : types of t1 columns are determined by all parts of union 
+
+create view v1 as
+with
+t1 as
+(
+select a  from t2 where t2.a=3
+union
+select t2.a+1 from t1,t2 where t1.a=t2.a
+)
+select * from t1;
+
+show columns from v1;
+
+
+--echo # WITH RECURSIVE : types of t1 columns are determined by anchor parts 
+
+create view v2 as
+with recursive
+t1 as
+(
+select a  from t2 where t2.a=3
+union
+select t2.a+1 from t1,t2 where t1.a=t2.a
+)
+select * from t1;
+
+show columns from v2;
+
+drop view v1,v2;
+
+drop table t1,t2;
+
+
 create table  folks(id int, name char(32), dob date, father int, mother int);
 
 insert into folks values
@@ -381,7 +480,6 @@ select generation, name from ancestor_ids a, folks
   where a.id = folks.id;
 
 set standards_compliant_cte=0;
-set optimizer_switch='materialization=off,subquery_cache=off';
 
 --ERROR ER_WITH_COL_WRONG_LIST
 with recursive
@@ -420,7 +518,6 @@ as
 select generation, name from ancestor_ids a, folks
   where a.id = folks.id;
 
-set optimizer_switch=default;
 set standards_compliant_cte=1;
 
 --ERROR ER_NOT_STANDARDS_COMPLIANT_RECURSIVE
@@ -585,5 +682,28 @@ as
 )
 select * from ancestors;
 
+set statement max_recursion_level=2 for
+with recursive
+ancestor_ids (id, generation)
+as
+(
+  select father, 1 from folks where name = 'Me'
+  union
+  select mother, 1 from folks where name = 'Me'
+  union
+  select father, a.generation+1 from folks, ancestor_ids a
+    where folks.id = a.id
+  union
+  select mother, a.generation+1 from folks, ancestor_ids a
+    where folks.id = a.id
+),
+ancestors 
+as
+(
+  select generation, name from folks as p, ancestor_ids as a
+    where p.id = a.id
+)
+select * from ancestors;
+
 drop table folks;
 

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);
 }
 
 

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};

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;
    }

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

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;

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

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 */

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,

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);
 

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,

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 "