Fixed numerous problems for mutually recursive CTE.

Actually mutually recursive CTE were not functional. Now the code
for mutually recursive CTE looks like functional, but still needs
re-writing.
Added many new test cases for mutually recursive CTE.

8 files changed, 653 insertions, 32 deletions

diff --git a/mysql-test/r/cte_recursive.result b/mysql-test/r/cte_recursive.result
index c2a820f6e34..2d8fac6269d 100644
--- a/mysql-test/r/cte_recursive.result
+++ b/mysql-test/r/cte_recursive.result
@@ -372,9 +372,9 @@ id	name	dob	father	mother
 100	Me	2000-01-01	20	30
 20	Dad	1970-02-02	10	9
 30	Mom	1975-03-03	8	7
-9	Grandma Ann	1941-10-15	NULL	NULL
 10	Grandpa Bill	1940-04-05	NULL	NULL
 8	Grandpa Ben	1940-10-21	NULL	NULL
+9	Grandma Ann	1941-10-15	NULL	NULL
 7	Grandma Sally	1943-08-23	NULL	6
 6	Grandgrandma Martha	1923-05-17	NULL	NULL
 with recursive
@@ -406,6 +406,254 @@ h_name	h_dob	w_name	w_dob
 Dad	1970-02-02	Mom	1975-03-03
 Grandpa Bill	1940-04-05	Grandma Ann	1941-10-15
 Grandpa Ben	1940-10-21	Grandma Sally	1943-08-23
+with recursive
+ancestor_couples(h_id, h_name, h_dob, h_father, h_mother,
+w_id, w_name, w_dob, w_father, w_mother)
+as
+(
+select h.*, w.*
+from folks h, folks w,  coupled_ancestors a
+where a.father = h.id AND a.mother = w.id
+union 
+select h.*, w.*
+from folks v, folks h, folks w
+where v.name = 'Me' and
+(v.father = h.id AND v.mother= w.id)
+),
+coupled_ancestors (id, name, dob, father, mother)
+as
+(
+select h_id, h_name, h_dob, h_father, h_mother
+from ancestor_couples
+union all
+select w_id, w_name, w_dob, w_father, w_mother
+from ancestor_couples
+)
+select h_name, h_dob, w_name, w_dob
+from ancestor_couples;
+h_name	h_dob	w_name	w_dob
+Dad	1970-02-02	Mom	1975-03-03
+Grandpa Bill	1940-04-05	Grandma Ann	1941-10-15
+Grandpa Ben	1940-10-21	Grandma Sally	1943-08-23
+with recursive
+ancestor_couples(h_id, h_name, h_dob, h_father, h_mother,
+w_id, w_name, w_dob, w_father, w_mother)
+as
+(
+select h.*, w.*
+from folks h, folks w,  coupled_ancestors a
+where a.father = h.id AND a.mother = w.id
+),
+coupled_ancestors (id, name, dob, father, mother)
+as
+(
+select * 
+from folks 
+where name = 'Me'
+  union all
+select h_id, h_name, h_dob, h_father, h_mother
+from ancestor_couples
+union all
+select w_id, w_name, w_dob, w_father, w_mother
+from ancestor_couples
+)
+select h_name, h_dob, w_name, w_dob
+from ancestor_couples;
+h_name	h_dob	w_name	w_dob
+Dad	1970-02-02	Mom	1975-03-03
+Grandpa Bill	1940-04-05	Grandma Ann	1941-10-15
+Grandpa Ben	1940-10-21	Grandma Sally	1943-08-23
+with recursive
+ancestor_couple_ids(h_id, w_id)
+as
+(
+select a.father, a.mother
+from coupled_ancestors a
+where a.father is not null and a.mother is not null
+),
+coupled_ancestors (id, name, dob, father, mother)
+as
+(
+select * 
+from folks 
+where name = 'Me'
+  union all
+select p.*
+from folks p, ancestor_couple_ids fa
+where p.id = fa.h_id
+union all
+select p.*
+from folks p, ancestor_couple_ids ma
+where p.id = ma.w_id
+)
+select *
+from ancestor_couple_ids;
+h_id	w_id
+20	30
+10	9
+8	7
+with recursive
+ancestor_couple_ids(h_id, w_id)
+as
+(
+select a.father, a.mother
+from coupled_ancestors a
+where a.father is not null and a.mother is not null
+),
+coupled_ancestors (id, name, dob, father, mother)
+as
+(
+select * 
+from folks 
+where name = 'Me'
+  union all
+select p.*
+from folks p, ancestor_couple_ids fa
+where p.id = fa.h_id
+union all
+select p.*
+from folks p, ancestor_couple_ids ma
+where p.id = ma.w_id
+)
+select h.name, h.dob, w.name, w.dob
+from ancestor_couple_ids c, folks h, folks w
+where c.h_id = h.id and c.w_id= w.id;
+name	dob	name	dob
+Dad	1970-02-02	Mom	1975-03-03
+Grandpa Bill	1940-04-05	Grandma Ann	1941-10-15
+Grandpa Ben	1940-10-21	Grandma Sally	1943-08-23
+with recursive
+ancestor_couple_ids(h_id, w_id)
+as
+(
+select a.father, a.mother
+from coupled_ancestors a
+where a.father is not null and a.mother is not null
+),
+coupled_ancestors (id, name, dob, father, mother)
+as
+(
+select * 
+from folks 
+where name = 'Me'
+  union all
+select p.*
+from folks p, ancestor_couple_ids fa
+where p.id = fa.h_id
+union all
+select p.*
+from folks p, ancestor_couple_ids ma
+where p.id = ma.w_id
+)
+select h.name, h.dob, w.name, w.dob
+from ancestor_couple_ids c, coupled_ancestors h, coupled_ancestors w
+where c.h_id = h.id and c.w_id= w.id;
+name	dob	name	dob
+Dad	1970-02-02	Mom	1975-03-03
+Grandpa Bill	1940-04-05	Grandma Ann	1941-10-15
+Grandpa Ben	1940-10-21	Grandma Sally	1943-08-23
+explain extended
+with recursive
+ancestor_couple_ids(h_id, w_id)
+as
+(
+select a.father, a.mother
+from coupled_ancestors a
+where a.father is not null and a.mother is not null
+),
+coupled_ancestors (id, name, dob, father, mother)
+as
+(
+select * 
+from folks 
+where name = 'Me'
+  union all
+select p.*
+from folks p, ancestor_couple_ids fa
+where p.id = fa.h_id
+union all
+select p.*
+from folks p, ancestor_couple_ids ma
+where p.id = ma.w_id
+)
+select h.name, h.dob, w.name, w.dob
+from ancestor_couple_ids c, coupled_ancestors h, coupled_ancestors w
+where c.h_id = h.id and c.w_id= w.id;
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	filtered	Extra
+1	PRIMARY	<derived2>	ALL	NULL	NULL	NULL	NULL	36	100.00	
+1	PRIMARY	<derived3>	ALL	NULL	NULL	NULL	NULL	468	100.00	Using where; Using join buffer (flat, BNL join)
+1	PRIMARY	<derived3>	ALL	NULL	NULL	NULL	NULL	468	100.00	Using where; Using join buffer (incremental, BNL join)
+3	SUBQUERY	folks	ALL	NULL	NULL	NULL	NULL	12	100.00	Using where
+4	UNCACHEABLE UNION	p	ALL	NULL	NULL	NULL	NULL	12	100.00	
+4	UNCACHEABLE UNION	<derived2>	ALL	NULL	NULL	NULL	NULL	36	100.00	Using where; Using join buffer (flat, BNL join)
+5	UNCACHEABLE UNION	<derived2>	ALL	NULL	NULL	NULL	NULL	2	100.00	
+5	UNCACHEABLE UNION	p	ALL	NULL	NULL	NULL	NULL	12	100.00	Using where; Using join buffer (flat, BNL join)
+NULL	UNION RESULT	<union3,4,5>	ALL	NULL	NULL	NULL	NULL	NULL	NULL	
+2	UNCACHEABLE SUBQUERY	<derived3>	ALL	NULL	NULL	NULL	NULL	36	100.00	Using where
+Warnings:
+Note	1003	with recursive ancestor_couple_ids as (select `a`.`father` AS `h_id`,`a`.`mother` AS `w_id` from `coupled_ancestors` `a` where ((`a`.`father` is not null) and (`a`.`mother` is not null)))coupled_ancestors as (select `test`.`folks`.`id` AS `id`,`test`.`folks`.`name` AS `name`,`test`.`folks`.`dob` AS `dob`,`test`.`folks`.`father` AS `father`,`test`.`folks`.`mother` AS `mother` from `test`.`folks` where (`test`.`folks`.`name` = 'Me') union all select `test`.`p`.`id` AS `id`,`test`.`p`.`name` AS `name`,`test`.`p`.`dob` AS `dob`,`test`.`p`.`father` AS `father`,`test`.`p`.`mother` AS `mother` from `test`.`folks` `p` join `ancestor_couple_ids` `fa` where (`fa`.`h_id` = `test`.`p`.`id`) union all select `test`.`p`.`id` AS `id`,`test`.`p`.`name` AS `name`,`test`.`p`.`dob` AS `dob`,`test`.`p`.`father` AS `father`,`test`.`p`.`mother` AS `mother` from `test`.`folks` `p` join `ancestor_couple_ids` `ma` where (`test`.`p`.`id` = `ma`.`w_id`)), select `h`.`name` AS `name`,`h`.`dob` AS `dob`,`w`.`name` AS `name`,`w`.`dob` AS `dob` from `ancestor_couple_ids` `c` join `coupled_ancestors` `h` join `coupled_ancestors` `w` where ((`h`.`id` = `c`.`h_id`) and (`w`.`id` = `c`.`w_id`))
+with recursive
+ancestor_couple_ids(h_id, w_id)
+as
+(
+select a.father, a.mother
+from coupled_ancestors a
+),
+coupled_ancestors (id, name, dob, father, mother)
+as
+(
+select * 
+from folks 
+where name = 'Me'
+  union all
+select p.*
+from folks p, ancestor_couple_ids fa
+where p.id = fa.h_id
+union all
+select p.*
+from folks p, ancestor_couple_ids ma
+where p.id = ma.w_id
+)
+select *
+from ancestor_couple_ids;
+h_id	w_id
+20	30
+10	9
+8	7
+NULL	NULL
+NULL	NULL
+NULL	NULL
+NULL	6
+NULL	NULL
+with recursive
+ancestor_couple_ids(h_id, w_id)
+as
+(
+select a.father, a.mother
+from coupled_ancestors a
+),
+coupled_ancestors (id, name, dob, father, mother)
+as
+(
+select * 
+from folks 
+where name = 'Me'
+  union all
+select p.*
+from folks p, ancestor_couple_ids fa
+where p.id = fa.h_id
+union all
+select p.*
+from folks p, ancestor_couple_ids ma
+where p.id = ma.w_id
+)
+select h.name, h.dob, w.name, w.dob
+from ancestor_couple_ids c, coupled_ancestors h, coupled_ancestors w
+where c.h_id = h.id and c.w_id= w.id;
+name	dob	name	dob
+Dad	1970-02-02	Mom	1975-03-03
+Grandpa Bill	1940-04-05	Grandma Ann	1941-10-15
+Grandpa Ben	1940-10-21	Grandma Sally	1943-08-23
 prepare stmt1 from "
 with recursive 
 ancestors
@@ -495,9 +743,9 @@ id	name	dob	father	mother
 100	Me	2000-01-01	20	30
 20	Dad	1970-02-02	10	9
 30	Mom	1975-03-03	8	7
-9	Grandma Ann	1941-10-15	NULL	NULL
 10	Grandpa Bill	1940-04-05	NULL	NULL
 8	Grandpa Ben	1940-10-21	NULL	NULL
+9	Grandma Ann	1941-10-15	NULL	NULL
 7	Grandma Sally	1943-08-23	NULL	6
 6	Grandgrandma Martha	1923-05-17	NULL	NULL
 drop view v1,v2;
@@ -571,7 +819,6 @@ id	name	dob	father	mother
 100	Me	2000-01-01	20	30
 20	Dad	1970-02-02	10	9
 30	Mom	1975-03-03	8	7
-9	Grandma Ann	1941-10-15	NULL	NULL
 10	Grandpa Bill	1940-04-05	NULL	NULL
 8	Grandpa Ben	1940-10-21	NULL	NULL
 9	Grandma Ann	1941-10-15	NULL	NULL
diff --git a/mysql-test/t/cte_recursive.test b/mysql-test/t/cte_recursive.test
index 6fef24be34f..911c381d46a 100644
--- a/mysql-test/t/cte_recursive.test
+++ b/mysql-test/t/cte_recursive.test
@@ -338,6 +338,221 @@ select h_name, h_dob, w_name, w_dob
   from ancestor_couples;
 
 
+with recursive
+ancestor_couples(h_id, h_name, h_dob, h_father, h_mother,
+                 w_id, w_name, w_dob, w_father, w_mother)
+as
+(
+  select h.*, w.*
+    from folks h, folks w,  coupled_ancestors a
+      where a.father = h.id AND a.mother = w.id
+  union 
+  select h.*, w.*
+    from folks v, folks h, folks w
+      where v.name = 'Me' and
+            (v.father = h.id AND v.mother= w.id)
+),
+coupled_ancestors (id, name, dob, father, mother)
+as
+(
+  select h_id, h_name, h_dob, h_father, h_mother
+    from ancestor_couples
+  union all
+  select w_id, w_name, w_dob, w_father, w_mother
+    from ancestor_couples
+)
+select h_name, h_dob, w_name, w_dob
+  from ancestor_couples;
+
+
+with recursive
+ancestor_couples(h_id, h_name, h_dob, h_father, h_mother,
+                 w_id, w_name, w_dob, w_father, w_mother)
+as
+(
+  select h.*, w.*
+    from folks h, folks w,  coupled_ancestors a
+      where a.father = h.id AND a.mother = w.id
+),
+coupled_ancestors (id, name, dob, father, mother)
+as
+(
+  select * 
+    from folks 
+      where name = 'Me'
+  union all
+  select h_id, h_name, h_dob, h_father, h_mother
+    from ancestor_couples
+  union all
+  select w_id, w_name, w_dob, w_father, w_mother
+    from ancestor_couples
+)
+select h_name, h_dob, w_name, w_dob
+  from ancestor_couples;
+
+
+with recursive
+ancestor_couple_ids(h_id, w_id)
+as
+(
+  select a.father, a.mother
+    from coupled_ancestors a
+      where a.father is not null and a.mother is not null
+),
+coupled_ancestors (id, name, dob, father, mother)
+as
+(
+  select * 
+    from folks 
+      where name = 'Me'
+  union all
+  select p.*
+    from folks p, ancestor_couple_ids fa
+      where p.id = fa.h_id
+  union all
+  select p.*
+    from folks p, ancestor_couple_ids ma
+      where p.id = ma.w_id
+)
+select *
+  from ancestor_couple_ids;
+
+with recursive
+ancestor_couple_ids(h_id, w_id)
+as
+(
+  select a.father, a.mother
+    from coupled_ancestors a
+      where a.father is not null and a.mother is not null
+),
+coupled_ancestors (id, name, dob, father, mother)
+as
+(
+  select * 
+    from folks 
+      where name = 'Me'
+  union all
+  select p.*
+    from folks p, ancestor_couple_ids fa
+      where p.id = fa.h_id
+  union all
+  select p.*
+    from folks p, ancestor_couple_ids ma
+      where p.id = ma.w_id
+)
+select h.name, h.dob, w.name, w.dob
+  from ancestor_couple_ids c, folks h, folks w
+    where c.h_id = h.id and c.w_id= w.id;
+
+
+with recursive
+ancestor_couple_ids(h_id, w_id)
+as
+(
+  select a.father, a.mother
+    from coupled_ancestors a
+      where a.father is not null and a.mother is not null
+),
+coupled_ancestors (id, name, dob, father, mother)
+as
+(
+  select * 
+    from folks 
+      where name = 'Me'
+  union all
+  select p.*
+    from folks p, ancestor_couple_ids fa
+      where p.id = fa.h_id
+  union all
+  select p.*
+    from folks p, ancestor_couple_ids ma
+      where p.id = ma.w_id
+)
+select h.name, h.dob, w.name, w.dob
+  from ancestor_couple_ids c, coupled_ancestors h, coupled_ancestors w
+    where c.h_id = h.id and c.w_id= w.id;
+
+explain extended
+with recursive
+ancestor_couple_ids(h_id, w_id)
+as
+(
+  select a.father, a.mother
+    from coupled_ancestors a
+      where a.father is not null and a.mother is not null
+),
+coupled_ancestors (id, name, dob, father, mother)
+as
+(
+  select * 
+    from folks 
+      where name = 'Me'
+  union all
+  select p.*
+    from folks p, ancestor_couple_ids fa
+      where p.id = fa.h_id
+  union all
+  select p.*
+    from folks p, ancestor_couple_ids ma
+      where p.id = ma.w_id
+)
+select h.name, h.dob, w.name, w.dob
+  from ancestor_couple_ids c, coupled_ancestors h, coupled_ancestors w
+    where c.h_id = h.id and c.w_id= w.id;
+
+with recursive
+ancestor_couple_ids(h_id, w_id)
+as
+(
+  select a.father, a.mother
+    from coupled_ancestors a
+),
+coupled_ancestors (id, name, dob, father, mother)
+as
+(
+  select * 
+    from folks 
+      where name = 'Me'
+  union all
+  select p.*
+    from folks p, ancestor_couple_ids fa
+      where p.id = fa.h_id
+  union all
+  select p.*
+    from folks p, ancestor_couple_ids ma
+      where p.id = ma.w_id
+)
+select *
+  from ancestor_couple_ids;
+
+
+with recursive
+ancestor_couple_ids(h_id, w_id)
+as
+(
+  select a.father, a.mother
+    from coupled_ancestors a
+),
+coupled_ancestors (id, name, dob, father, mother)
+as
+(
+  select * 
+    from folks 
+      where name = 'Me'
+  union all
+  select p.*
+    from folks p, ancestor_couple_ids fa
+      where p.id = fa.h_id
+  union all
+  select p.*
+    from folks p, ancestor_couple_ids ma
+      where p.id = ma.w_id
+)
+select h.name, h.dob, w.name, w.dob
+  from ancestor_couple_ids c, coupled_ancestors h, coupled_ancestors w
+    where c.h_id = h.id and c.w_id= w.id;
+
+
 prepare stmt1 from "
 with recursive 
 ancestors
diff --git a/sql/sql_cte.h b/sql/sql_cte.h
index 23eea8463e6..ed16b82d352 100644
--- a/sql/sql_cte.h
+++ b/sql/sql_cte.h
@@ -87,6 +87,7 @@ public:
 
   TABLE *result_table;
 
+
   With_element(LEX_STRING *name,
                List <LEX_STRING> list,
                st_select_lex_unit *unit)
@@ -164,6 +165,30 @@ public:
 
   void reset_for_exec();
 
+  bool no_driving_recursive_is_set();
+
+  void set_as_driving_recursive();
+
+  bool is_driving_recursive();
+
+  void cleanup_driving_recursive();
+
+  void cleanup_incr_ready();
+
+  void set_as_incr_ready();
+
+  bool is_incr_ready();
+
+  bool all_incr_are_ready();
+
+  void cleanup_stabilized();
+
+  void set_as_stabilized();
+
+  bool is_stabilized();
+
+  bool all_are_stabilized();
+
   void set_result_table(TABLE *tab) { result_table= tab; }
 
   friend class With_clause;
@@ -202,6 +227,9 @@ private:
   table_map unrestricted;
   table_map with_prepared_anchor;
   table_map cleaned;
+  table_map driving_recursive;
+  table_map incr_ready;
+  table_map stabilized;
 
 public:
  /* If true the specifier RECURSIVE is present in the with clause */
@@ -211,7 +239,8 @@ public:
     : owner(NULL), first_elem(NULL), elements(0),
       embedding_with_clause(emb_with_clause), next_with_clause(NULL),
       dependencies_are_checked(false),
-    unrestricted(0), with_prepared_anchor(0), cleaned(0),
+      unrestricted(0), with_prepared_anchor(0), cleaned(0),
+      driving_recursive(0), incr_ready(0), stabilized(0),
       with_recursive(recursive_fl)
   { last_next= &first_elem; }
 
@@ -287,6 +316,7 @@ bool With_element::is_cleaned()
   return owner->cleaned & get_elem_map();
 }
 
+
 inline
 void With_element::mark_as_cleaned() 
 {
@@ -299,9 +329,97 @@ void With_element::reset_for_exec()
 {
   level= 0;
   owner->with_prepared_anchor&= ~mutually_recursive;
-  owner->cleaned&= ~get_elem_map();    
+  owner->cleaned&= ~get_elem_map();
+  owner->driving_recursive&= ~get_elem_map();
+  cleanup_incr_ready();
+  cleanup_stabilized();
+}
+
+
+inline
+bool With_element::no_driving_recursive_is_set()
+{
+  return !(owner->driving_recursive & mutually_recursive);
+}
+
+
+inline
+void With_element::set_as_driving_recursive()
+{
+  owner->driving_recursive|= get_elem_map();
+}
+
+
+inline
+bool With_element::is_driving_recursive()
+{
+  return owner->driving_recursive & get_elem_map();
+}
+
+
+inline
+void With_element::cleanup_driving_recursive()
+{
+  owner->driving_recursive&= ~mutually_recursive;
+}
+
+
+inline
+void With_element::cleanup_incr_ready()
+{
+  owner->incr_ready&= ~mutually_recursive;
+}
+
+
+inline
+void With_element::set_as_incr_ready()
+{
+  owner->incr_ready|= get_elem_map();
+}
+
+
+inline
+bool With_element::is_incr_ready()
+{
+  return owner->incr_ready & get_elem_map();
+}
+
+
+inline
+bool With_element::all_incr_are_ready()
+{
+  return (owner->incr_ready & mutually_recursive) == mutually_recursive;
+}
+
+
+inline
+void With_element::cleanup_stabilized()
+{
+  owner->stabilized&= ~mutually_recursive;
+}
+
+
+inline
+void With_element::set_as_stabilized()
+{
+  owner->stabilized|= get_elem_map();
 }
 
+
+inline
+bool With_element::is_stabilized()
+{
+  return owner->stabilized & get_elem_map();
+}
+
+
+inline
+bool With_element::all_are_stabilized()
+{
+  return (owner->stabilized & mutually_recursive) == mutually_recursive;
+}
+
+
 inline
 void  st_select_lex_unit::set_with_clause(With_clause *with_cl)
 { 
@@ -310,6 +428,7 @@ void  st_select_lex_unit::set_with_clause(With_clause *with_cl)
       with_clause->set_owner(this);
 }
 
+
 inline
 void st_select_lex::set_with_clause(With_clause *with_clause)
 {
diff --git a/sql/sql_derived.cc b/sql/sql_derived.cc
index 0db94421aa5..f84cdd939fe 100644
--- a/sql/sql_derived.cc
+++ b/sql/sql_derived.cc
@@ -661,7 +661,10 @@ bool mysql_derived_prepare(THD *thd, LEX *lex, TABLE_LIST *derived)
       derived->derived_result->set_unit(unit);
       derived->table= derived->derived_result->table;
       if (derived->is_with_table_recursive_reference())
+      {
         unit->with_element->rec_result->rec_tables.push_back(derived->table);
+        derived->table->is_rec_table= true;
+      }
     }
     DBUG_ASSERT(derived->table || res);
     goto exit;
@@ -945,9 +948,10 @@ bool mysql_derived_fill(THD *thd, LEX *lex, TABLE_LIST *derived)
 {
   DBUG_ENTER("mysql_derived_fill");
   SELECT_LEX_UNIT *unit= derived->get_unit();
+  bool derived_is_recursive= derived->is_recursive_with_table();
   bool res= FALSE;
 
-  if (derived->is_recursive_with_table() && unit->executed)
+  if (derived_is_recursive && derived->with->all_are_stabilized())
   {
     TABLE *src= unit->with_element->rec_result->table;
     TABLE *dest= derived->table;
@@ -955,17 +959,18 @@ bool mysql_derived_fill(THD *thd, LEX *lex, TABLE_LIST *derived)
     DBUG_RETURN(res);
   }
 
-  if (unit->executed && !unit->uncacheable && !unit->describe)
+  if (unit->executed && !unit->uncacheable && !unit->describe &&
+      !derived_is_recursive)
     DBUG_RETURN(FALSE);
   /*check that table creation passed without problems. */
   DBUG_ASSERT(derived->table && derived->table->is_created());
   SELECT_LEX *first_select= unit->first_select();
   select_union *derived_result= derived->derived_result;
   SELECT_LEX *save_current_select= lex->current_select;
-  if (unit->is_union())
+  if (unit->is_union() || derived_is_recursive)
   {
     // execute union without clean up
-    if (derived->is_recursive_with_table())
+    if (derived_is_recursive)
       unit->with_element->set_result_table(derived->table);
     res= unit->exec();
   }
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 84dd2e4b676..25a509472dc 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -11605,7 +11605,12 @@ bool JOIN_TAB::preread_init()
   }
 
   /* Materialize derived table/view. */
-  if ((!derived->get_unit()->executed  || derived->is_recursive_with_table()) &&
+  if ((!derived->get_unit()->executed  ||
+       (derived->is_recursive_with_table() &&
+        (!derived->is_with_table_recursive_reference() ||
+         (!derived->with->is_driving_recursive() &&
+          !derived->with->is_incr_ready()) &&
+	 !derived->with->all_are_stabilized()))) &&
       mysql_handle_single_derived(join->thd->lex,
                                     derived, DT_CREATE | DT_FILL))
       return TRUE;
@@ -18236,7 +18241,8 @@ sub_select(JOIN *join,JOIN_TAB *join_tab,bool end_of_records)
     flush_dups_table->sj_weedout_delete_rows();
   }
 
-  if (!join_tab->preread_init_done && join_tab->preread_init())
+  if ((!join_tab->preread_init_done || join_tab->table->is_rec_table) &&
+      join_tab->preread_init())
     DBUG_RETURN(NESTED_LOOP_ERROR);
 
   join->return_tab= join_tab;
@@ -19189,7 +19195,8 @@ int join_init_read_record(JOIN_TAB *tab)
     report_error(tab->table, error);
     return 1;
   }
-  if (!tab->preread_init_done && tab->preread_init())
+  if ((!tab->preread_init_done || tab->table->is_rec_table) &&
+      tab->preread_init())
     return 1;
   if (init_read_record(&tab->read_record, tab->join->thd, tab->table,
                        tab->select, tab->filesort_result, 1,1, FALSE))
diff --git a/sql/sql_union.cc b/sql/sql_union.cc
index d19bbaf103c..384cca70574 100644
--- a/sql/sql_union.cc
+++ b/sql/sql_union.cc
@@ -244,6 +244,7 @@ select_union_recursive::create_result_table(THD *thd_arg,
 
   if (rec_tables.push_back(rec_table))
     return true;
+  rec_table->is_rec_table= true;
 
   return false;
 }
@@ -494,7 +495,7 @@ bool st_select_lex_unit::prepare(THD *thd_arg, select_result *sel_result,
 
   /* Global option */
 
-  if (is_union_select)
+  if (is_union_select || is_recursive)
   {
     if (is_union() && !union_needs_tmp_table())
     {
@@ -530,7 +531,7 @@ bool st_select_lex_unit::prepare(THD *thd_arg, select_result *sel_result,
   sl->context.resolve_in_select_list= TRUE;
  
   for (;sl; sl= sl->next_select())
-  {
+  {  
     bool can_skip_order_by;
     sl->options|=  SELECT_NO_UNLOCK;
     JOIN *join= new JOIN(thd_arg, sl->item_list, 
@@ -587,7 +588,7 @@ bool st_select_lex_unit::prepare(THD *thd_arg, select_result *sel_result,
       Use items list of underlaid select for derived tables to preserve
       information about fields lengths and exact types
     */
-    if (!is_union_select)
+    if (!is_union_select && !is_recursive)
       types= first_sl->item_list;
     else if (sl == first_sl)
     {
@@ -917,8 +918,9 @@ bool st_select_lex_unit::exec()
   bool first_execution= !executed;
   DBUG_ENTER("st_select_lex_unit::exec");
   bool was_executed= executed;
+  bool is_recursive= with_element && with_element->is_recursive;
 
-  if (executed && !uncacheable && !describe)
+  if (executed && !uncacheable && !describe && !is_recursive)
     DBUG_RETURN(FALSE);
   executed= 1;
   if (!(uncacheable & ~UNCACHEABLE_EXPLAIN) && item)
@@ -934,7 +936,7 @@ bool st_select_lex_unit::exec()
   if (saved_error)
     DBUG_RETURN(saved_error);
 
-  if (with_element && with_element->is_recursive && !describe)
+  if (is_recursive && !describe)
   {
     saved_error= exec_recursive();
     DBUG_RETURN(saved_error);
@@ -1174,10 +1176,7 @@ bool st_select_lex_unit::exec_recursive()
   TABLE *result_table= with_element->result_table;
   ha_rows examined_rows= 0;
   bool unrestricted= with_element->is_unrestricted();
-  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;
@@ -1186,9 +1185,31 @@ bool st_select_lex_unit::exec_recursive()
 
   do
   {
+    st_select_lex *first_sl;
+    st_select_lex *barrier;
     if ((saved_error= incr_table->file->ha_delete_all_rows()))
       goto err;
 
+    if (with_element->no_driving_recursive_is_set())
+        with_element->set_as_driving_recursive();
+
+    if (with_element->level == 0)
+    {
+      first_sl= first_select();
+      if (with_anchor)
+        barrier= first_recursive_sel;
+      else
+	barrier= NULL;
+    }
+    else
+    {
+      first_sl= first_recursive_sel;
+      barrier= NULL;
+    }
+
+    if (with_element->all_incr_are_ready())
+      with_element->cleanup_incr_ready();
+   
     for (st_select_lex *sl= first_sl ; sl != barrier; sl= sl->next_select())
     {
       thd->lex->current_select= sl;
@@ -1211,16 +1232,12 @@ bool st_select_lex_unit::exec_recursive()
       }
     }
 
-    if (with_element->level == 0)
-    {
-      first_sl= first_recursive_sel;
-      barrier= NULL;
-    }
+    with_element->set_as_incr_ready();
 
     incr_table->file->info(HA_STATUS_VARIABLE);
     if (incr_table->file->stats.records == 0 ||
         with_element->level + 1 == max_level)
-       no_more_iterations= true;
+      with_element->set_as_stabilized();
     else
       with_element->level++;
 
@@ -1231,12 +1248,21 @@ bool st_select_lex_unit::exec_recursive()
                                                          !unrestricted)))
 	  goto err;
     }
-  } while (!no_more_iterations); 
+    
+    if (!with_element->is_driving_recursive())
+      break;
+
+  } while (!with_element->all_are_stabilized()); 
 
-  if ((saved_error= table->insert_all_rows_into(thd,
-                                                result_table,
-                                                true)))
-    goto err;
+  if (with_element->is_driving_recursive())
+  {
+    TABLE *table= with_element->rec_result->table;
+    if ((saved_error= table->insert_all_rows_into(thd,
+                                                  result_table,
+                                                  true)))
+      goto err;
+    with_element->cleanup_driving_recursive();
+  }
 
   thd->lex->current_select= lex_select_save;
 err:
diff --git a/sql/table.cc b/sql/table.cc
index 6109c16fb37..e11ea59bcf8 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -7328,7 +7328,8 @@ bool TABLE_LIST::init_derived(THD *thd, bool init_view)
     if (!is_materialized_derived() && first_select->is_mergeable() &&
         optimizer_flag(thd, OPTIMIZER_SWITCH_DERIVED_MERGE) &&
         !(thd->lex->sql_command == SQLCOM_UPDATE_MULTI ||
-          thd->lex->sql_command == SQLCOM_DELETE_MULTI))
+          thd->lex->sql_command == SQLCOM_DELETE_MULTI) &&
+        !is_recursive_with_table())
       set_merged_derived();
     else
       set_materialized_derived();
diff --git a/sql/table.h b/sql/table.h
index 122b036cae5..a8d01a64599 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -1244,6 +1244,7 @@ public:
   bool alias_name_used;              /* true if table_name is alias */
   bool get_fields_in_item_tree;      /* Signal to fix_field */
   bool m_needs_reopen;
+  bool is_rec_table;
 private:
   bool created;    /* For tmp tables. TRUE <=> tmp table was actually created.*/
 public: