20 files changed, 617 insertions, 145 deletions

diff --git a/include/mysql_com.h b/include/mysql_com.h
index 4eafe148743..cc89f014f8e 100644
--- a/include/mysql_com.h
+++ b/include/mysql_com.h
@@ -206,6 +206,8 @@ enum enum_indicator_type
                                                 itself supports it*/
 #define LONG_UNIQUE_HASH_FIELD       (1<< 30) /* This field will store hash for unique
                                                 column */
+#define FIELD_PART_OF_TMP_UNIQUE     (1<< 31) /* part of an unique constrain
+                                                for a tmporary table*/
 
 #define REFRESH_GRANT           (1ULL << 0)  /* Refresh grant tables */
 #define REFRESH_LOG             (1ULL << 1)  /* Start on new log file */
diff --git a/mysql-test/main/brackets.result b/mysql-test/main/brackets.result
index 548250db758..3de7d902335 100644
--- a/mysql-test/main/brackets.result
+++ b/mysql-test/main/brackets.result
@@ -4454,7 +4454,7 @@ create view v1 as
 with t(a) as (values (2), (1)) select a from t;
 show create view v1;
 View	Create View	character_set_client	collation_connection
-v1	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `v1` AS with t(a) as (values (2),(1))select `t`.`a` AS `a` from `t`	latin1	latin1_swedish_ci
+v1	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `v1` AS with t(`a`) as (values (2),(1))select `t`.`a` AS `a` from `t`	latin1	latin1_swedish_ci
 select * from v1;
 a
 2
@@ -4469,7 +4469,7 @@ order by 1 desc limit 3 )
 select a from t1 where a=4 union select a from t where a=7 order by a desc;
 show create view v1;
 View	Create View	character_set_client	collation_connection
-v1	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `v1` AS with t(a) as (values (2),(1) union (values (4),(7)) order by 1 desc limit 3)select `t1`.`a` AS `a` from `t1` where `t1`.`a` = 4 union select `t`.`a` AS `a` from `t` where `t`.`a` = 7 order by `a` desc	latin1	latin1_swedish_ci
+v1	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `v1` AS with t(`a`) as (values (2),(1) union (values (4),(7)) order by 1 desc limit 3)select `t1`.`a` AS `a` from `t1` where `t1`.`a` = 4 union select `t`.`a` AS `a` from `t` where `t`.`a` = 7 order by `a` desc	latin1	latin1_swedish_ci
 select * from v1;
 a
 7
@@ -4484,7 +4484,7 @@ order by 1 desc limit 3 )
 select a from t1 where a=1 union select a from t where a=7 order by a desc;
 show create view v1;
 View	Create View	character_set_client	collation_connection
-v1	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `v1` AS with t(a) as ((values (2),(1)) union (values (4),(7) order by 1 desc) order by 1 desc limit 3)select `t1`.`a` AS `a` from `t1` where `t1`.`a` = 1 union select `t`.`a` AS `a` from `t` where `t`.`a` = 7 order by `a` desc	latin1	latin1_swedish_ci
+v1	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `v1` AS with t(`a`) as ((values (2),(1)) union (values (4),(7) order by 1 desc) order by 1 desc limit 3)select `t1`.`a` AS `a` from `t1` where `t1`.`a` = 1 union select `t`.`a` AS `a` from `t` where `t`.`a` = 7 order by `a` desc	latin1	latin1_swedish_ci
 select * from v1;
 a
 7
diff --git a/mysql-test/main/cte_cycle.result b/mysql-test/main/cte_cycle.result
new file mode 100644
index 00000000000..e66d090b3c7
--- /dev/null
+++ b/mysql-test/main/cte_cycle.result
@@ -0,0 +1,156 @@
+#
+# check errors
+#
+WITH RECURSIVE cte AS (
+SELECT 1 AS a UNION ALL
+SELECT NULL FROM cte WHERE a IS NOT NULL)
+CYCLE a, a RESTRICT
+SELECT * FROM cte;
+ERROR 42S21: Duplicate column name 'a'
+WITH RECURSIVE cte AS (
+SELECT 1 AS a UNION ALL
+SELECT NULL FROM cte WHERE a IS NOT NULL)
+CYCLE b RESTRICT
+SELECT * FROM cte;
+ERROR 42S22: Unknown column 'b' in 'CYCLE clause'
+WITH cte AS (
+SELECT 1 AS a UNION ALL
+SELECT NULL FROM cte WHERE a IS NOT NULL)
+CYCLE b RESTRICT
+SELECT * FROM cte;
+ERROR 42000: You have an error in your SQL syntax; check the manual that corresponds to your MariaDB server version for the right syntax to use near 'CYCLE b RESTRICT
+SELECT * FROM cte' at line 4
+#
+# A degenerate case
+#
+WITH RECURSIVE cte AS (
+SELECT 1 AS a, 2 as b)
+CYCLE b RESTRICT
+SELECT * FROM cte;
+a	b
+1	2
+#
+# A simple case
+#
+WITH RECURSIVE cte AS (
+SELECT 1 AS a, 2 as b UNION ALL
+SELECT 2, 2 FROM cte WHERE a IS NOT NULL)
+CYCLE b RESTRICT
+SELECT * FROM cte;
+a	b
+1	2
+#
+# MDEV-20632 case (with fixed syntax)
+#
+create table t1 (from_ int, to_ int);
+insert into t1 values (1,2), (1,100), (2,3), (3,4), (4,1);
+WITH RECURSIVE cte (depth, from_, to_) as (
+SELECT 0,1,1
+UNION
+SELECT depth+1, t1.from_, t1.to_
+FROM t1, cte WHERE t1.from_ = cte.to_
+) CYCLE from_, to_ RESTRICT
+select * from cte;
+depth	from_	to_
+0	1	1
+1	1	2
+1	1	100
+2	2	3
+3	3	4
+4	4	1
+create view v1 as WITH RECURSIVE cte (depth, from_, to_) as (
+SELECT 0,1,1
+UNION
+SELECT depth+1, t1.from_, t1.to_
+FROM t1, cte WHERE t1.from_ = cte.to_
+) CYCLE from_, to_ RESTRICT
+select * from cte;
+show create view v1;
+View	Create View	character_set_client	collation_connection
+v1	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `v1` AS with recursive cte(`depth`,`from_`,`to_`) as (select 0 AS `depth`,1 AS `from_`,1 AS `to_` union select `cte`.`depth` + 1 AS `depth+1`,`t1`.`from_` AS `from_`,`t1`.`to_` AS `to_` from (`t1` join `cte`) where `t1`.`from_` = `cte`.`to_`) CYCLE `from_`,`to_` RESTRICT select `cte`.`depth` AS `depth`,`cte`.`from_` AS `from_`,`cte`.`to_` AS `to_` from `cte`	latin1	latin1_swedish_ci
+select * from v1;
+depth	from_	to_
+0	1	1
+1	1	2
+1	1	100
+2	2	3
+3	3	4
+4	4	1
+delete from t1;
+insert into t1 values (1,2), (1,NULL), (NULL,NULL), (NULL, 1);
+select * from v1;
+depth	from_	to_
+0	1	1
+1	1	2
+1	1	NULL
+drop view v1;
+drop table t1;
+#
+# A simple blob case
+#
+create table t1 (a int, b text);
+insert into t1 values (1, "a");
+WITH RECURSIVE cte AS (
+SELECT a, b  from t1 UNION ALL
+SELECT a, b FROM cte WHERE a IS NOT NULL)
+CYCLE b RESTRICT
+SELECT * FROM cte;
+a	b
+1	a
+drop table t1;
+#
+# check bit types
+#
+create table t1 (from_ bit(3), to_ bit(3));
+insert into t1 values (1,2), (1,7), (2,3), (3,4), (4,1);
+WITH RECURSIVE cte (depth, from_, to_) as (
+SELECT 0,1,1
+UNION
+SELECT depth+1, t1.from_, t1.to_
+FROM t1, cte WHERE t1.from_ = cte.to_
+) CYCLE from_, to_ RESTRICT
+select * from cte;
+depth	from_	to_
+0	1	1
+1	1	2
+1	1	7
+2	2	3
+3	3	4
+4	4	1
+drop table t1;
+#
+# check bit types with BLOBs (TEXT)
+#
+create table t1 (from_ bit(3), to_ bit(3), load_ text);
+insert into t1 values (1,2,"A"), (1,7,"A"), (2,3,"A"), (3,4,"A"), (4,1,"A");
+WITH RECURSIVE cte (depth, from_, to_, load_) as (
+SELECT 0,1,1,"A"
+   UNION
+SELECT depth+1, t1.from_, t1.to_, t1.load_
+FROM t1, cte WHERE t1.from_ = cte.to_
+) CYCLE from_, to_, load_ RESTRICT
+select * from cte;
+depth	from_	to_	load_
+0	1	1	A
+1	1	2	A
+1	1	7	A
+2	2	3	A
+3	3	4	A
+4	4	1	A
+insert into t1 values (4,1,"B");
+WITH RECURSIVE cte (depth, from_, to_, load_) as (
+SELECT 0,1,1,"A"
+   UNION
+SELECT depth+1, t1.from_, t1.to_, t1.load_
+FROM t1, cte WHERE t1.from_ = cte.to_
+) CYCLE from_, to_, load_ RESTRICT
+select * from cte;
+depth	from_	to_	load_
+0	1	1	A
+1	1	2	A
+1	1	7	A
+2	2	3	A
+3	3	4	A
+4	4	1	A
+4	4	1	B
+drop table t1;
diff --git a/mysql-test/main/cte_cycle.test b/mysql-test/main/cte_cycle.test
new file mode 100644
index 00000000000..505980353fd
--- /dev/null
+++ b/mysql-test/main/cte_cycle.test
@@ -0,0 +1,143 @@
+
+--echo #
+--echo # check errors
+--echo #
+
+--error ER_DUP_FIELDNAME
+WITH RECURSIVE cte AS (
+        SELECT 1 AS a UNION ALL
+        SELECT NULL FROM cte WHERE a IS NOT NULL)
+        CYCLE a, a RESTRICT
+SELECT * FROM cte;
+
+--error ER_BAD_FIELD_ERROR
+WITH RECURSIVE cte AS (
+        SELECT 1 AS a UNION ALL
+        SELECT NULL FROM cte WHERE a IS NOT NULL)
+        CYCLE b RESTRICT
+SELECT * FROM cte;
+
+--error ER_PARSE_ERROR
+WITH cte AS (
+        SELECT 1 AS a UNION ALL
+        SELECT NULL FROM cte WHERE a IS NOT NULL)
+        CYCLE b RESTRICT
+SELECT * FROM cte;
+
+
+--echo #
+--echo # A degenerate case
+--echo #
+
+WITH RECURSIVE cte AS (
+	SELECT 1 AS a, 2 as b)
+   CYCLE b RESTRICT
+SELECT * FROM cte;
+
+
+--echo #
+--echo # A simple case
+--echo #
+
+WITH RECURSIVE cte AS (
+	SELECT 1 AS a, 2 as b UNION ALL
+	SELECT 2, 2 FROM cte WHERE a IS NOT NULL)
+        CYCLE b RESTRICT
+SELECT * FROM cte;
+
+
+--echo #
+--echo # MDEV-20632 case (with fixed syntax)
+--echo #
+
+create table t1 (from_ int, to_ int);
+insert into t1 values (1,2), (1,100), (2,3), (3,4), (4,1);
+
+WITH RECURSIVE cte (depth, from_, to_) as (
+   SELECT 0,1,1
+   UNION
+   SELECT depth+1, t1.from_, t1.to_
+      FROM t1, cte WHERE t1.from_ = cte.to_
+) CYCLE from_, to_ RESTRICT
+select * from cte;
+
+create view v1 as WITH RECURSIVE cte (depth, from_, to_) as (
+   SELECT 0,1,1
+   UNION
+   SELECT depth+1, t1.from_, t1.to_
+      FROM t1, cte WHERE t1.from_ = cte.to_
+) CYCLE from_, to_ RESTRICT
+select * from cte;
+
+show create view v1;
+
+select * from v1;
+
+delete from t1;
+
+insert into t1 values (1,2), (1,NULL), (NULL,NULL), (NULL, 1);
+
+select * from v1;
+
+drop view v1;
+
+drop table t1;
+
+
+--echo #
+--echo # A simple blob case
+--echo #
+
+create table t1 (a int, b text);
+insert into t1 values (1, "a");
+
+WITH RECURSIVE cte AS (
+	SELECT a, b  from t1 UNION ALL
+	SELECT a, b FROM cte WHERE a IS NOT NULL)
+        CYCLE b RESTRICT
+SELECT * FROM cte;
+
+drop table t1;
+
+--echo #
+--echo # check bit types
+--echo #
+
+create table t1 (from_ bit(3), to_ bit(3));
+insert into t1 values (1,2), (1,7), (2,3), (3,4), (4,1);
+
+WITH RECURSIVE cte (depth, from_, to_) as (
+   SELECT 0,1,1
+   UNION
+   SELECT depth+1, t1.from_, t1.to_
+      FROM t1, cte WHERE t1.from_ = cte.to_
+) CYCLE from_, to_ RESTRICT
+select * from cte;
+drop table t1;
+
+--echo #
+--echo # check bit types with BLOBs (TEXT)
+--echo #
+
+create table t1 (from_ bit(3), to_ bit(3), load_ text);
+insert into t1 values (1,2,"A"), (1,7,"A"), (2,3,"A"), (3,4,"A"), (4,1,"A");
+
+WITH RECURSIVE cte (depth, from_, to_, load_) as (
+   SELECT 0,1,1,"A"
+   UNION
+   SELECT depth+1, t1.from_, t1.to_, t1.load_
+      FROM t1, cte WHERE t1.from_ = cte.to_
+) CYCLE from_, to_, load_ RESTRICT
+select * from cte;
+insert into t1 values (4,1,"B");
+WITH RECURSIVE cte (depth, from_, to_, load_) as (
+   SELECT 0,1,1,"A"
+   UNION
+   SELECT depth+1, t1.from_, t1.to_, t1.load_
+      FROM t1, cte WHERE t1.from_ = cte.to_
+) CYCLE from_, to_, load_ RESTRICT
+select * from cte;
+
+drop table t1;
+
+
diff --git a/mysql-test/main/cte_nonrecursive.result b/mysql-test/main/cte_nonrecursive.result
index 53d9d5ecc8f..a24ebdd1fff 100644
--- a/mysql-test/main/cte_nonrecursive.result
+++ b/mysql-test/main/cte_nonrecursive.result
@@ -606,7 +606,7 @@ with t(c) as (select a from t1 where b >= 'c')
 select * from t r1 where r1.c=4;
 show create view v3;
 View	Create View	character_set_client	collation_connection
-v3	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `v3` AS with t(c) as (select `t1`.`a` AS `c` from `t1` where `t1`.`b` >= 'c')select `r1`.`c` AS `c` from `t` `r1` where `r1`.`c` = 4	latin1	latin1_swedish_ci
+v3	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `v3` AS with t(`c`) as (select `t1`.`a` AS `c` from `t1` where `t1`.`b` >= 'c')select `r1`.`c` AS `c` from `t` `r1` where `r1`.`c` = 4	latin1	latin1_swedish_ci
 select * from v3;
 c
 4
@@ -618,7 +618,7 @@ with t(c) as (select a from t1 where b >= 'c')
 select * from t r1, t r2 where r1.c=r2.c and r2.c=4;
 show create view v4;
 View	Create View	character_set_client	collation_connection
-v4	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `v4` AS with t(c) as (select `test`.`t1`.`a` AS `c` from `test`.`t1` where `test`.`t1`.`b` >= 'c')select `r1`.`c` AS `c`,`r2`.`c` AS `d` from (`t` `r1` join (select `test`.`t1`.`a` AS `c` from `test`.`t1` where `test`.`t1`.`b` >= 'c') `r2`) where `r1`.`c` = `r2`.`c` and `r2`.`c` = 4	latin1	latin1_swedish_ci
+v4	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `v4` AS with t(`c`) as (select `test`.`t1`.`a` AS `c` from `test`.`t1` where `test`.`t1`.`b` >= 'c')select `r1`.`c` AS `c`,`r2`.`c` AS `d` from (`t` `r1` join (select `test`.`t1`.`a` AS `c` from `test`.`t1` where `test`.`t1`.`b` >= 'c') `r2`) where `r1`.`c` = `r2`.`c` and `r2`.`c` = 4	latin1	latin1_swedish_ci
 select * from v4;
 c	d
 4	4
diff --git a/mysql-test/main/cte_recursive.result b/mysql-test/main/cte_recursive.result
index d439c2fc39b..2f00248fb55 100644
--- a/mysql-test/main/cte_recursive.result
+++ b/mysql-test/main/cte_recursive.result
@@ -699,7 +699,7 @@ id	select_type	table	type	possible_keys	key	key_len	ref	rows	filtered	Extra
 5	RECURSIVE 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	
 Warnings:
-Note	1003	with recursive ancestor_couple_ids(h_id,w_id) as (/* select#2 */ 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(id,name,dob,father,mother) as (/* select#3 */ 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#4 */ 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 `test`.`p`.`id` = `fa`.`h_id` union all /* select#5 */ 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#1 */ 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`
+Note	1003	with recursive ancestor_couple_ids(`h_id`,`w_id`) as (/* select#2 */ 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(`id`,`name`,`dob`,`father`,`mother`) as (/* select#3 */ 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#4 */ 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 `test`.`p`.`id` = `fa`.`h_id` union all /* select#5 */ 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#1 */ 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`
 # simple mutual recursion
 with recursive
 ancestor_couple_ids(h_id, w_id)
@@ -3093,7 +3093,7 @@ id	select_type	table	type	possible_keys	key	key_len	ref	rows	filtered	Extra
 4	DEPENDENT SUBQUERY	<derived2>	ALL	NULL	NULL	NULL	NULL	16	100.00	Using where
 NULL	UNION RESULT	<union2,3>	ALL	NULL	NULL	NULL	NULL	NULL	NULL	
 Warnings:
-Note	1003	with recursive destinations(city,legs) as (/* select#2 */ select `test`.`a`.`arrival` AS `city`,1 AS `legs` from `test`.`flights` `a` where `test`.`a`.`departure` = 'Cairo' union /* select#3 */ select `test`.`b`.`arrival` AS `arrival`,`r`.`legs` + 1 AS `r.legs + 1` from `destinations` `r` join `test`.`flights` `b` where `r`.`city` = `test`.`b`.`departure` and !<in_optimizer>(`test`.`b`.`arrival`,<exists>(/* select#4 */ select `destinations`.`city` from `destinations` where trigcond(`test`.`b`.`arrival` = `destinations`.`city` or `destinations`.`city` is null) having trigcond(`destinations`.`city` is null))))/* select#1 */ select `destinations`.`city` AS `city`,`destinations`.`legs` AS `legs` from `destinations`
+Note	1003	with recursive destinations(`city`,`legs`) as (/* select#2 */ select `test`.`a`.`arrival` AS `city`,1 AS `legs` from `test`.`flights` `a` where `test`.`a`.`departure` = 'Cairo' union /* select#3 */ select `test`.`b`.`arrival` AS `arrival`,`r`.`legs` + 1 AS `r.legs + 1` from `destinations` `r` join `test`.`flights` `b` where `r`.`city` = `test`.`b`.`departure` and !<in_optimizer>(`test`.`b`.`arrival`,<exists>(/* select#4 */ select `destinations`.`city` from `destinations` where trigcond(`test`.`b`.`arrival` = `destinations`.`city` or `destinations`.`city` is null) having trigcond(`destinations`.`city` is null))))/* select#1 */ select `destinations`.`city` AS `city`,`destinations`.`legs` AS `legs` from `destinations`
 set standard_compliant_cte=default;
 drop table flights;
 #
@@ -3380,7 +3380,7 @@ id	select_type	table	type	possible_keys	key	key_len	ref	rows	filtered	Extra
 3	RECURSIVE UNION	<derived2>	ALL	NULL	NULL	NULL	NULL	2	100.00	Using where
 NULL	UNION RESULT	<union2,3>	ALL	NULL	NULL	NULL	NULL	NULL	NULL	
 Warnings:
-Note	1003	with recursive rcte(a) as (/* select#2 */ select 1 AS `a` union /* select#3 */ select cast(`rcte`.`a` + 1 as unsigned) AS `cast(a+1 as unsigned)` from `rcte` where `rcte`.`a` < 10), cte1 as (/* select#4 */ select count(0) AS `c1` from `rcte` join `test`.`t1` where `rcte`.`a` between 3 and 5 and `test`.`t1`.`id` = `rcte`.`a` - 3), cte2 as (/* select#5 */ select count(0) AS `c2` from `rcte` join `test`.`t1` where `rcte`.`a` between 7 and 8 and `test`.`t1`.`id` = `rcte`.`a` - 7)/* select#1 */ select `cte1`.`c1` AS `c1`,`cte2`.`c2` AS `c2` from `cte1` join `cte2`
+Note	1003	with recursive rcte(`a`) as (/* select#2 */ select 1 AS `a` union /* select#3 */ select cast(`rcte`.`a` + 1 as unsigned) AS `cast(a+1 as unsigned)` from `rcte` where `rcte`.`a` < 10), cte1 as (/* select#4 */ select count(0) AS `c1` from `rcte` join `test`.`t1` where `rcte`.`a` between 3 and 5 and `test`.`t1`.`id` = `rcte`.`a` - 3), cte2 as (/* select#5 */ select count(0) AS `c2` from `rcte` join `test`.`t1` where `rcte`.`a` between 7 and 8 and `test`.`t1`.`id` = `rcte`.`a` - 7)/* select#1 */ select `cte1`.`c1` AS `c1`,`cte2`.`c2` AS `c2` from `cte1` join `cte2`
 prepare stmt from "with recursive
 rcte(a) as
 (select 1 union select cast(a+1 as unsigned) from rcte where a < 10),
diff --git a/mysql-test/main/table_value_constr.result b/mysql-test/main/table_value_constr.result
index 54bf9c36491..38e1a54082c 100644
--- a/mysql-test/main/table_value_constr.result
+++ b/mysql-test/main/table_value_constr.result
@@ -2593,7 +2593,7 @@ ERROR 42S22: Unknown column '2' in 'order clause'
 create view v1 as with t(a) as (values (2), (1)) select a from t;
 show create view v1;
 View	Create View	character_set_client	collation_connection
-v1	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `v1` AS with t(a) as (values (2),(1))select `t`.`a` AS `a` from `t`	latin1	latin1_swedish_ci
+v1	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `v1` AS with t(`a`) as (values (2),(1))select `t`.`a` AS `a` from `t`	latin1	latin1_swedish_ci
 select * from v1;
 a
 2
diff --git a/sql/item.cc b/sql/item.cc
index 6b585325439..49005030930 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -403,7 +403,7 @@ int Item::save_str_value_in_field(Field *field, String *result)
 
 Item::Item(THD *thd):
   is_expensive_cache(-1), rsize(0), name(null_clex_str), orig_name(0),
-  is_autogenerated_name(TRUE)
+  common_flags(IS_AUTO_GENERATED_NAME)
 {
   DBUG_ASSERT(thd);
   marker= 0;
@@ -463,7 +463,7 @@ Item::Item(THD *thd, Item *item):
   with_param(item->with_param),
   with_window_func(item->with_window_func),
   with_field(item->with_field),
-  is_autogenerated_name(item->is_autogenerated_name)
+  common_flags(item->common_flags)
 {
   next= thd->free_list;				// Put in free list
   thd->free_list= this;
@@ -1117,7 +1117,7 @@ void Item::set_name(THD *thd, const char *str, size_t length, CHARSET_INFO *cs)
       str++;
     }
   }
-  if (str != str_start && !is_autogenerated_name)
+  if (str != str_start && !is_autogenerated_name())
   {
     char buff[SAFE_NAME_LEN];
 
@@ -5099,7 +5099,7 @@ static Item** find_field_in_group_list(Item *find_item, ORDER *group_list)
 
     /* SELECT list element with explicit alias */
     if ((*(cur_group->item))->name.str && !table_name.str &&
-        !(*(cur_group->item))->is_autogenerated_name &&
+        !(*(cur_group->item))->is_autogenerated_name() &&
         !lex_string_cmp(system_charset_info,
                         &(*(cur_group->item))->name, &field_name))
     {
diff --git a/sql/item.h b/sql/item.h
index 6a9d401b101..c51ea35b254 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -624,6 +624,13 @@ class st_select_lex_unit;
 class Item_func_not;
 class Item_splocal;
 
+/* Item::common_flags */
+/* Indicates that name of this Item autogenerated or set by user */
+#define IS_AUTO_GENERATED_NAME 1
+/* Indicates that this item is in CYCLE clause of WITH */
+#define IS_IN_WITH_CYCLE       2
+
+
 /**
   String_copier that sends Item specific warnings.
 */
@@ -931,8 +938,9 @@ public:
     True if any item except Item_sum contains a field. Set during parsing.
   */
   bool with_field;
-  bool is_autogenerated_name;           /* indicate was name of this Item
-                                           autogenerated or set by user */
+  uint8 common_flags;
+  bool is_autogenerated_name()
+  { return (common_flags & IS_AUTO_GENERATED_NAME); }
   // alloc & destruct is done as start of select on THD::mem_root
   Item(THD *thd);
   /*
@@ -7028,6 +7036,7 @@ public:
     name= item->name;
     Type_std_attributes::set(*attr);
     maybe_null= maybe_null_arg;
+    common_flags= item->common_flags;
   }
 
   const Type_handler *type_handler() const
diff --git a/sql/item_create.cc b/sql/item_create.cc
index 57e402999ea..2bd605286d0 100644
--- a/sql/item_create.cc
+++ b/sql/item_create.cc
@@ -2387,7 +2387,7 @@ static bool has_named_parameters(List<Item> *params)
     List_iterator<Item> it(*params);
     while ((param= it++))
     {
-      if (! param->is_autogenerated_name)
+      if (! param->is_autogenerated_name())
         return true;
     }
   }
@@ -2633,7 +2633,7 @@ Create_func_arg1::create_func(THD *thd, LEX_CSTRING *name, List<Item> *item_list
 
   Item *param_1= item_list->pop();
 
-  if (unlikely(! param_1->is_autogenerated_name))
+  if (unlikely(! param_1->is_autogenerated_name()))
   {
     my_error(ER_WRONG_PARAMETERS_TO_NATIVE_FCT, MYF(0), name->str);
     return NULL;
@@ -2660,8 +2660,8 @@ Create_func_arg2::create_func(THD *thd, LEX_CSTRING *name, List<Item> *item_list
   Item *param_1= item_list->pop();
   Item *param_2= item_list->pop();
 
-  if (unlikely(!param_1->is_autogenerated_name ||
-               !param_2->is_autogenerated_name))
+  if (unlikely(!param_1->is_autogenerated_name() ||
+               !param_2->is_autogenerated_name()))
   {
     my_error(ER_WRONG_PARAMETERS_TO_NATIVE_FCT, MYF(0), name->str);
     return NULL;
@@ -2689,9 +2689,9 @@ Create_func_arg3::create_func(THD *thd, LEX_CSTRING *name, List<Item> *item_list
   Item *param_2= item_list->pop();
   Item *param_3= item_list->pop();
 
-  if (unlikely(!param_1->is_autogenerated_name ||
-               !param_2->is_autogenerated_name ||
-               !param_3->is_autogenerated_name))
+  if (unlikely(!param_1->is_autogenerated_name() ||
+               !param_2->is_autogenerated_name() ||
+               !param_3->is_autogenerated_name()))
   {
     my_error(ER_WRONG_PARAMETERS_TO_NATIVE_FCT, MYF(0), name->str);
     return NULL;
diff --git a/sql/item_subselect.cc b/sql/item_subselect.cc
index 2d3ea388cc5..4b781152c33 100644
--- a/sql/item_subselect.cc
+++ b/sql/item_subselect.cc
@@ -4434,9 +4434,10 @@ void subselect_single_select_engine::print(String *str,
                                            enum_query_type query_type)
 {
   With_clause* with_clause= select_lex->get_with_clause();
+  THD *thd= get_thd();
   if (with_clause)
-    with_clause->print(str, query_type);
-  select_lex->print(get_thd(), str, query_type);
+    with_clause->print(thd, str, query_type);
+  select_lex->print(thd, str, query_type);
 }
 
 
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index c88aef09c96..3d49ebc31c8 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -5767,7 +5767,7 @@ find_field_in_view(THD *thd, TABLE_LIST *table_list,
        replace. If the item was aliased by the user, set the alias to
        the replacing item.
       */
-      if (*ref && !(*ref)->is_autogenerated_name)
+      if (*ref && !(*ref)->is_autogenerated_name())
         item->set_name(thd, (*ref)->name);
       if (register_tree_change)
         thd->change_item_tree(ref, item);
@@ -5858,7 +5858,7 @@ find_field_in_natural_join(THD *thd, TABLE_LIST *table_ref, const char *name, si
      replace. If the item was aliased by the user, set the alias to
      the replacing item.
      */
-    if (*ref && !(*ref)->is_autogenerated_name)
+    if (*ref && !(*ref)->is_autogenerated_name())
       item->set_name(thd, (*ref)->name);
     if (register_tree_change && arena)
       thd->restore_active_arena(arena, &backup);
diff --git a/sql/sql_cte.cc b/sql/sql_cte.cc
index c7a0f9186e2..93344956468 100644
--- a/sql/sql_cte.cc
+++ b/sql/sql_cte.cc
@@ -21,6 +21,7 @@
 #include "sql_view.h"    // for make_valid_column_names
 #include "sql_parse.h"
 #include "sql_select.h"
+#include "sql_show.h"                // append_definer, append_identifier
 
 
 /**
@@ -944,9 +945,9 @@ err:
     false  otherwise
 */    
 
-bool 
-With_element::rename_columns_of_derived_unit(THD *thd, 
-                                             st_select_lex_unit *unit)
+bool
+With_element::process_columns_of_derived_unit(THD *thd,
+                                              st_select_lex_unit *unit)
 {
   if (unit->columns_are_renamed)
     return false;
@@ -973,7 +974,7 @@ With_element::rename_columns_of_derived_unit(THD *thd,
     while ((item= it++, name= nm++))
     {
       item->set_name(thd, *name);
-      item->is_autogenerated_name= false;
+      item->common_flags&= ~IS_AUTO_GENERATED_NAME;
     }
 
     if (arena)
@@ -982,6 +983,43 @@ With_element::rename_columns_of_derived_unit(THD *thd,
   else
     make_valid_column_names(thd, select->item_list);
 
+  if (cycle_list)
+  {
+    List_iterator_fast<Item> it(select->item_list);
+    List_iterator_fast<Lex_ident_sys> nm(*cycle_list);
+    List_iterator_fast<Lex_ident_sys> nm_check(*cycle_list);
+    DBUG_ASSERT(cycle_list->elements != 0);
+    while (LEX_CSTRING *name= nm++)
+    {
+      Item *item;
+      /*
+        Check for uniqueness of each element in the cycle list:
+        It's sufficient to check that there is no duplicate of 'name'
+        among the elements that precede it.
+      */
+      LEX_CSTRING *check;
+      nm_check.rewind();
+      while ((check= nm_check++) && check != name)
+      {
+        if (check->length == name->length &&
+            strncmp(check->str, name->str, name->length) == 0)
+        {
+          my_error(ER_DUP_FIELDNAME, MYF(0), check->str);
+          return true;
+        }
+      }
+      /* Check that 'name' is the name of a column of the processed CTE */
+      while ((item= it++) &&
+             (item->name.length != name->length ||
+              strncmp(item->name.str, name->str, name->length) != 0));
+      if (item == NULL)
+      {
+        my_error(ER_BAD_FIELD_ERROR, MYF(0), name->str, "CYCLE clause");
+        return true;
+      }
+      item->common_flags|= IS_IN_WITH_CYCLE;
+    }
+  }
   unit->columns_are_renamed= true;
 
   return false;
@@ -1018,7 +1056,7 @@ bool With_element::prepare_unreferenced(THD *thd)
   thd->lex->context_analysis_only|= CONTEXT_ANALYSIS_ONLY_DERIVED;
   if (!spec->prepared &&
       (spec->prepare(spec->derived, 0, 0) ||
-       rename_columns_of_derived_unit(thd, spec) ||
+       process_columns_of_derived_unit(thd, spec) ||
        check_duplicate_names(thd, first_sl->item_list, 1)))
     rc= true;
 
@@ -1394,16 +1432,17 @@ bool st_select_lex::check_subqueries_with_recursive_references()
 /**
   @brief
     Print this with clause
-  
+
+  @param thd         Thread handle
   @param str         Where to print to
-  @param query_type  The mode of printing 
-     
+  @param query_type  The mode of printing
+
   @details
-    The method prints a string representation of this clause in the 
+    The method prints a string representation of this clause in the
     string str. The parameter query_type specifies the mode of printing.
-*/    
+*/
 
-void With_clause::print(String *str, enum_query_type query_type)
+void With_clause::print(THD *thd, String *str, enum_query_type query_type)
 {
   /*
     Any with clause contains just definitions of CTE tables.
@@ -1420,7 +1459,22 @@ void With_clause::print(String *str, enum_query_type query_type)
   {
     if (with_elem != with_list.first)
       str->append(", ");
-    with_elem->print(str, query_type);
+    with_elem->print(thd, str, query_type);
+  }
+}
+
+
+static void list_strlex_print(THD *thd, String *str, List<Lex_ident_sys> *list)
+{
+  List_iterator_fast<Lex_ident_sys> li(*list);
+  bool first= TRUE;
+  while(Lex_ident_sys *col_name= li++)
+  {
+    if (first)
+      first= FALSE;
+    else
+      str->append(',');
+    append_identifier(thd, str, col_name);
   }
 }
 
@@ -1428,38 +1482,37 @@ void With_clause::print(String *str, enum_query_type query_type)
 /**
    @brief
      Print this with element
-  
+
+  @param thd         Thread handle
   @param str         Where to print to
-  @param query_type  The mode of printing 
-     
+  @param query_type  The mode of printing
+
   @details
-    The method prints a string representation of this with element in the 
+    The method prints a string representation of this with element in the
     string str. The parameter query_type specifies the mode of printing.
 */
 
-void With_element::print(String *str, enum_query_type query_type)
+void With_element::print(THD *thd, String *str, enum_query_type query_type)
 {
   str->append(query_name);
   if (column_list.elements)
   {
     List_iterator_fast<Lex_ident_sys> li(column_list);
     str->append('(');
-    for (LEX_CSTRING *col_name= li++; ; )
-    {
-      str->append(col_name);
-      col_name= li++;
-      if (!col_name)
-      {
-        str->append(')');
-        break;
-      }
-      str->append(',');
-    }
+    list_strlex_print(thd, str, &column_list);
+    str->append(')');
   }
-  str->append(STRING_WITH_LEN(" as "));
-  str->append('(');
+  str->append(STRING_WITH_LEN(" as ("));
   spec->print(str, query_type);
   str->append(')');
+
+  if (cycle_list)
+  {
+    DBUG_ASSERT(cycle_list->elements != 0);
+    str->append(STRING_WITH_LEN(" CYCLE "));
+    list_strlex_print(thd, str, cycle_list);
+    str->append(STRING_WITH_LEN(" RESTRICT "));
+  }
 }
 
 
@@ -1483,3 +1536,26 @@ bool With_element::instantiate_tmp_tables()
   return false;
 }
 
+void With_element::set_cycle_list(List<Lex_ident_sys> *cycle_list_arg)
+{
+  cycle_list= cycle_list_arg;
+
+  /*
+    If a CTE table with columns c1,...,cn is defined with a cycle
+    clause CYCLE(ci1,...,cik) then no two rows r1 and r2 from the
+    table shall have r1.ci1=r2.ci1 && ... && r1.cik=r2.cik.
+
+    If a cycle clause is used in the specification of a CTE then
+    each UNION ALL at the top level of the specification is interpreted
+    as a UNION DISTINCT over the cycle columns.
+  */
+  for (st_select_lex *sl=  spec->first_select(); sl; sl= sl->next_select())
+  {
+    spec->union_distinct= sl;
+    if (sl != spec->first_select())
+    {
+      sl->distinct= TRUE;
+      sl->with_all_modifier= FALSE;
+    }
+  }
+}
diff --git a/sql/sql_cte.h b/sql/sql_cte.h
index 92aca5090f3..4c42dd23614 100644
--- a/sql/sql_cte.h
+++ b/sql/sql_cte.h
@@ -112,6 +112,7 @@ public:
     always empty.
   */
   List <Lex_ident_sys> column_list;
+  List <Lex_ident_sys> *cycle_list;
   /* The query that specifies the table introduced by this with element */
   st_select_lex_unit *spec;
   /* 
@@ -169,7 +170,7 @@ public:
       sq_dep_map(0), work_dep_map(0), mutually_recursive(0),
       top_level_dep_map(0), sq_rec_ref(NULL),
       next_mutually_recursive(NULL), references(0), 
-      query_name(name), column_list(list), spec(unit),
+      query_name(name), column_list(list), cycle_list(0), spec(unit),
       is_recursive(false), rec_outer_references(0), with_anchor(false),
       level(0), rec_result(NULL)
   { unit->with_element= this; }
@@ -206,7 +207,7 @@ public:
 
   void inc_references() { references++; }
 
-  bool rename_columns_of_derived_unit(THD *thd, st_select_lex_unit *unit);
+  bool process_columns_of_derived_unit(THD *thd, st_select_lex_unit *unit);
 
   bool prepare_unreferenced(THD *thd);
 
@@ -214,7 +215,7 @@ public:
                                     table_map &unrestricted,
                                     table_map &encountered);
 
-  void print(String *str, enum_query_type query_type);
+  void print(THD *thd, String *str, enum_query_type query_type);
 
   With_clause *get_owner() { return owner; }
 
@@ -259,6 +260,8 @@ public:
 
   void prepare_for_next_iteration();
 
+  void set_cycle_list(List<Lex_ident_sys> *cycle_list_arg);
+
   friend class With_clause;
 };
 
@@ -353,7 +356,7 @@ public:
 
   void add_unrestricted(table_map map) { unrestricted|= map; }
 
-  void print(String *str, enum_query_type query_type);
+  void print(THD *thd, String *str, enum_query_type query_type);
 
   friend class With_element;
 
diff --git a/sql/sql_derived.cc b/sql/sql_derived.cc
index c713f1322dc..7d09c85a7c7 100644
--- a/sql/sql_derived.cc
+++ b/sql/sql_derived.cc
@@ -816,8 +816,8 @@ bool mysql_derived_prepare(THD *thd, LEX *lex, TABLE_LIST *derived)
   if ((res= unit->prepare(derived, derived->derived_result, 0)))
     goto exit;
   if (derived->with &&
-      (res= derived->with->rename_columns_of_derived_unit(thd, unit)))
-    goto exit; 
+      (res= derived->with->process_columns_of_derived_unit(thd, unit)))
+    goto exit;
   lex->context_analysis_only&= ~CONTEXT_ANALYSIS_ONLY_DERIVED;
   if ((res= check_duplicate_names(thd, unit->types, 0)))
     goto exit;
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index 12b266508d0..d448cd0e59e 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -3581,7 +3581,7 @@ void LEX::print(String *str, enum_query_type query_type)
 void st_select_lex_unit::print(String *str, enum_query_type query_type)
 {
   if (with_clause)
-    with_clause->print(str, query_type);
+    with_clause->print(thd, str, query_type);
   for (SELECT_LEX *sl= first_select(); sl; sl= sl->next_select())
   {
     if (sl != first_select())
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 09a15aaf7df..afd6ccb3cd3 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -18023,14 +18023,27 @@ class Create_tmp_table: public Data_type_statistics
   ORDER *m_group;
   bool m_distinct;
   bool m_save_sum_fields;
+  bool m_with_cycle;
   ulonglong m_select_options;
   ha_rows m_rows_limit;
-  uint m_hidden_field_count;       // Remove this eventually
   uint m_group_null_items;
-  uint m_null_count;
-  uint m_hidden_uneven_bit_length;
-  uint m_hidden_null_count;
+
+  // counter for distinct/other fields
+  uint m_field_count[2];
+  // counter for distinct/other fields which can be NULL
+  uint m_null_count[2];
+  // counter for distinct/other  blob fields
+  uint m_blobs_count[2];
+  // counter for "tails" of bit fields which do not fit in a byte
+  uint m_uneven_bit[2];
+
 public:
+  enum counter {distinct, other};
+  /*
+    shows which field we are processing: distinct/other (set in processing
+    cycles)
+  */
+  counter current_counter;
   Create_tmp_table(const TMP_TABLE_PARAM *param,
                    ORDER *group, bool distinct, bool save_sum_fields,
                    ulonglong select_options, ha_rows rows_limit)
@@ -18040,14 +18053,21 @@ public:
     m_group(group),
     m_distinct(distinct),
     m_save_sum_fields(save_sum_fields),
+    m_with_cycle(false),
     m_select_options(select_options),
     m_rows_limit(rows_limit),
-    m_hidden_field_count(param->hidden_field_count),
     m_group_null_items(0),
-    m_null_count(0),
-    m_hidden_uneven_bit_length(0),
-    m_hidden_null_count(0)
-  { }
+    current_counter(other)
+  {
+    m_field_count[Create_tmp_table::distinct]= 0;
+    m_field_count[Create_tmp_table::other]= 0;
+    m_null_count[Create_tmp_table::distinct]= 0;
+    m_null_count[Create_tmp_table::other]= 0;
+    m_blobs_count[Create_tmp_table::distinct]= 0;
+    m_blobs_count[Create_tmp_table::other]= 0;
+    m_uneven_bit[Create_tmp_table::distinct]= 0;
+    m_uneven_bit[Create_tmp_table::other]= 0;
+  }
 
   void add_field(TABLE *table, Field *field, uint fieldnr, bool force_not_null_cols);
 
@@ -18080,13 +18100,16 @@ void Create_tmp_table::add_field(TABLE *table, Field *field, uint fieldnr, bool
   }
 
   if (!(field->flags & NOT_NULL_FLAG))
-    m_null_count++;
+    m_null_count[current_counter]++;
 
   table->s->reclength+= field->pack_length();
 
   // Assign it here, before update_data_type_statistics() changes m_blob_count
   if (field->flags & BLOB_FLAG)
+  {
     table->s->blob_field[m_blob_count]= fieldnr;
+    m_blobs_count[current_counter]++;
+  }
 
   table->field[fieldnr]= field;
   field->field_index= fieldnr;
@@ -18298,6 +18321,7 @@ bool Create_tmp_table::add_fields(THD *thd,
   DBUG_ASSERT(table->s->blob_fields == 0);
 
   const bool not_all_columns= !(m_select_options & TMP_TABLE_ALL_COLUMNS);
+  bool distinct_record_structure= m_distinct;
   uint fieldnr= 0;
   TABLE_SHARE  *share= table->s;
   Item **copy_func= param->items_to_copy;
@@ -18308,8 +18332,29 @@ bool Create_tmp_table::add_fields(THD *thd,
   List_iterator_fast<Item> li(fields);
   Item *item;
   Field **tmp_from_field= m_from_field;
+  uint uneven_delta;
+  while (!m_with_cycle && (item= li++))
+    if (item->common_flags & IS_IN_WITH_CYCLE)
+    {
+      m_with_cycle= true;
+      /*
+        Following distinct_record_structure is (m_distinct || m_with_cycle)
+
+        Note: distinct_record_structure can be true even if m_distinct is
+        false, for example for incr_table in recursive CTE
+        (see select_union_recursive::create_result_table)
+      */
+      distinct_record_structure= true;
+    }
+  li.rewind();
   while ((item=li++))
   {
+    current_counter= (((param->hidden_field_count < (fieldnr + 1)) &&
+                       distinct_record_structure &&
+                       (!m_with_cycle ||
+                        (item->common_flags & IS_IN_WITH_CYCLE)))?
+                      distinct :
+                      other);
     Item::Type type= item->type();
     if (type == Item::COPY_STR_ITEM)
     {
@@ -18334,7 +18379,8 @@ bool Create_tmp_table::add_fields(THD *thd,
 	  continue;
         }
       }
-      if (item->const_item() && (int) m_hidden_field_count <= 0)
+      if (item->const_item() &&
+          param->hidden_field_count < (fieldnr + 1))
         continue; // We don't have to store this
     }
     if (type == Item::SUM_FUNC_ITEM && !m_group && !m_save_sum_fields)
@@ -18351,7 +18397,7 @@ bool Create_tmp_table::add_fields(THD *thd,
             create_tmp_field(table, arg, &copy_func,
                              tmp_from_field, &m_default_field[fieldnr],
                              m_group != 0, not_all_columns,
-                             m_distinct, false);
+                             distinct_record_structure , false);
 	  if (!new_field)
 	    goto err;					// Should be OOM
 	  tmp_from_field++;
@@ -18363,7 +18409,10 @@ bool Create_tmp_table::add_fields(THD *thd,
           arg= sum_item->set_arg(i, thd, tmp_item);
           thd->mem_root= &table->mem_root;
 
+          uneven_delta= m_uneven_bit_length;
           add_field(table, new_field, fieldnr++, param->force_not_null_cols);
+          uneven_delta= m_uneven_bit_length - uneven_delta;
+          m_field_count[current_counter]++;
 
 	  if (!(new_field->flags & NOT_NULL_FLAG))
           {
@@ -18373,6 +18422,8 @@ bool Create_tmp_table::add_fields(THD *thd,
             */
             arg->maybe_null=1;
           }
+          if (current_counter == distinct)
+            new_field->flags|= FIELD_PART_OF_TMP_UNIQUE;
 	}
       }
     }
@@ -18440,36 +18491,23 @@ bool Create_tmp_table::add_fields(THD *thd,
       }
       tmp_from_field++;
 
+      uneven_delta= m_uneven_bit_length;
       add_field(table, new_field, fieldnr++, param->force_not_null_cols);
+      uneven_delta= m_uneven_bit_length - uneven_delta;
+      m_field_count[current_counter]++;
 
       if (item->marker == 4 && item->maybe_null)
       {
         m_group_null_items++;
 	new_field->flags|= GROUP_FLAG;
       }
+      if (current_counter == distinct)
+        new_field->flags|= FIELD_PART_OF_TMP_UNIQUE;
     }
-    if (!--m_hidden_field_count)
-    {
-      /*
-        This was the last hidden field; Remember how many hidden fields could
-        have null
-      */
-      m_hidden_null_count= m_null_count;
-      /*
-	We need to update hidden_field_count as we may have stored group
-	functions with constant arguments
-      */
-      param->hidden_field_count= fieldnr;
-      m_null_count= 0;
-      /*
-        On last hidden field we store uneven bit length in
-        m_hidden_uneven_bit_length and proceed calculation of
-        uneven bits for visible fields into m_uneven_bit_length.
-      */
-      m_hidden_uneven_bit_length= m_uneven_bit_length;
-      m_uneven_bit_length= 0;
-    }
+    m_uneven_bit[current_counter]+= uneven_delta;
   }
+  DBUG_ASSERT(fieldnr == m_field_count[other] + m_field_count[distinct]);
+  DBUG_ASSERT(m_blob_count == m_blobs_count[other] + m_blobs_count[distinct]);
   share->fields= fieldnr;
   share->blob_fields= m_blob_count;
   table->field[fieldnr]= 0;                     // End marker
@@ -18495,8 +18533,12 @@ bool Create_tmp_table::finalize(THD *thd,
   DBUG_ENTER("Create_tmp_table::finalize");
   DBUG_ASSERT(table);
 
-  uint hidden_null_pack_length;
-  uint null_pack_length;
+  uint null_pack_length[2];
+  uint null_pack_base[2];
+  uint null_counter[2]= {0, 0};
+
+  uint whole_null_pack_length;
+
   bool  use_packed_rows= false;
   uchar *pos;
   uchar *null_flags;
@@ -18547,16 +18589,21 @@ bool Create_tmp_table::finalize(THD *thd,
   if (share->blob_fields == 0)
   {
     /* We need to ensure that first byte is not 0 for the delete link */
-    if (param->hidden_field_count)
-      m_hidden_null_count++;
+    if (m_field_count[other])
+      m_null_count[other]++;
     else
-      m_null_count++;
-  }
-  hidden_null_pack_length= (m_hidden_null_count + 7 +
-                            m_hidden_uneven_bit_length) / 8;
-  null_pack_length= (hidden_null_pack_length +
-                     (m_null_count + m_uneven_bit_length + 7) / 8);
-  share->reclength+= null_pack_length;
+      m_null_count[distinct]++;
+  }
+
+  null_pack_length[other]= (m_null_count[other] + 7 +
+                            m_uneven_bit[other]) / 8;
+  null_pack_base[other]= 0;
+  null_pack_length[distinct]= (m_null_count[distinct] + 7 +
+                              m_uneven_bit[distinct]) / 8;
+  null_pack_base[distinct]= null_pack_length[other];
+  whole_null_pack_length= null_pack_length[other] +
+                          null_pack_length[distinct];
+  share->reclength+= whole_null_pack_length;
   if (!share->reclength)
     share->reclength= 1;                // Dummy select
   /* Use packed rows if there is blobs or a lot of space to gain */
@@ -18580,43 +18627,53 @@ bool Create_tmp_table::finalize(THD *thd,
 
   recinfo=param->start_recinfo;
   null_flags=(uchar*) table->record[0];
-  pos=table->record[0]+ null_pack_length;
-  if (null_pack_length)
+  pos=table->record[0]+ whole_null_pack_length;
+  if (whole_null_pack_length)
   {
     bzero((uchar*) recinfo,sizeof(*recinfo));
     recinfo->type=FIELD_NORMAL;
-    recinfo->length=null_pack_length;
+    recinfo->length= whole_null_pack_length;
     recinfo++;
-    bfill(null_flags,null_pack_length,255);	// Set null fields
+    bfill(null_flags, whole_null_pack_length, 255);	// Set null fields
 
     table->null_flags= (uchar*) table->record[0];
-    share->null_fields= m_null_count + m_hidden_null_count;
-    share->null_bytes= share->null_bytes_for_compare= null_pack_length;
+    share->null_fields= m_null_count[other] + m_null_count[distinct];
+    share->null_bytes= share->null_bytes_for_compare= whole_null_pack_length;
+  }
+
+  if (share->blob_fields == 0)
+  {
+    null_counter[(m_field_count[other] ? other : distinct)]++;
   }
-  m_null_count= (share->blob_fields == 0) ? 1 : 0;
-  m_hidden_field_count= param->hidden_field_count;
   for (uint i= 0; i < share->fields; i++, recinfo++)
   {
     Field *field= table->field[i];
     uint length;
     bzero((uchar*) recinfo,sizeof(*recinfo));
 
+    current_counter= ((field->flags & FIELD_PART_OF_TMP_UNIQUE) ?
+                      distinct :
+                      other);
+
     if (!(field->flags & NOT_NULL_FLAG))
     {
-      recinfo->null_bit= (uint8)1 << (m_null_count & 7);
-      recinfo->null_pos= m_null_count/8;
-      field->move_field(pos, null_flags + m_null_count/8,
-			(uint8)1 << (m_null_count & 7));
-      m_null_count++;
+
+      recinfo->null_bit= (uint8)1 << (null_counter[current_counter] & 7);
+      recinfo->null_pos= (null_pack_base[current_counter] +
+                          null_counter[current_counter]/8);
+      field->move_field(pos, null_flags + recinfo->null_pos, recinfo->null_bit);
+      null_counter[current_counter]++;
     }
     else
       field->move_field(pos,(uchar*) 0,0);
     if (field->type() == MYSQL_TYPE_BIT)
     {
       /* We have to reserve place for extra bits among null bits */
-      ((Field_bit*) field)->set_bit_ptr(null_flags + m_null_count / 8,
-                                        m_null_count & 7);
-      m_null_count+= (field->field_length & 7);
+      ((Field_bit*) field)->set_bit_ptr(null_flags +
+                                        null_pack_base[current_counter] +
+                                        null_counter[current_counter]/8,
+                                        null_counter[current_counter] & 7);
+      null_counter[current_counter]+= (field->field_length & 7);
     }
     field->reset();
 
@@ -18655,8 +18712,6 @@ bool Create_tmp_table::finalize(THD *thd,
     /* Make entry for create table */
     recinfo->length=length;
     recinfo->type= field->tmp_engine_column_type(use_packed_rows);
-    if (!--m_hidden_field_count)
-      m_null_count= (m_null_count + 7) & ~7;    // move to next byte
 
     // fix table name in field entry
     field->set_table_name(&table->alias);
@@ -18777,7 +18832,8 @@ bool Create_tmp_table::finalize(THD *thd,
                 m_group_buff <= param->group_buff + param->group_length);
   }
 
-  if (m_distinct && share->fields != param->hidden_field_count)
+  if (m_distinct && (share->fields != param->hidden_field_count ||
+                     m_with_cycle))
   {
     uint i;
     Field **reg_field;
@@ -18789,7 +18845,7 @@ bool Create_tmp_table::finalize(THD *thd,
     */
     DBUG_PRINT("info",("hidden_field_count: %d", param->hidden_field_count));
 
-    if (share->blob_fields)
+    if (m_blobs_count[distinct])
     {
       /*
         Special mode for index creation in MyISAM used to support unique
@@ -18798,10 +18854,8 @@ bool Create_tmp_table::finalize(THD *thd,
       */
       share->uniques= 1;
     }
-    null_pack_length-=hidden_null_pack_length;
-    keyinfo->user_defined_key_parts=
-      ((share->fields - param->hidden_field_count)+
-       (share->uniques ? MY_TEST(null_pack_length) : 0));
+    keyinfo->user_defined_key_parts= m_field_count[distinct] +
+       (share->uniques ? MY_TEST(null_pack_length[distinct]) : 0);
     keyinfo->ext_key_parts= keyinfo->user_defined_key_parts;
     keyinfo->usable_key_parts= keyinfo->user_defined_key_parts;
     table->distinct= 1;
@@ -18844,11 +18898,11 @@ bool Create_tmp_table::finalize(THD *thd,
       blobs can distinguish NULL from 0. This extra field is not needed
       when we do not use UNIQUE indexes for blobs.
     */
-    if (null_pack_length && share->uniques)
+    if (null_pack_length[distinct] && share->uniques)
     {
       m_key_part_info->null_bit=0;
-      m_key_part_info->offset=hidden_null_pack_length;
-      m_key_part_info->length=null_pack_length;
+      m_key_part_info->offset= null_pack_base[distinct];
+      m_key_part_info->length= null_pack_length[distinct];
       m_key_part_info->field= new Field_string(table->record[0],
                                              (uint32) m_key_part_info->length,
                                              (uchar*) 0,
@@ -18866,8 +18920,10 @@ bool Create_tmp_table::finalize(THD *thd,
     /* Create a distinct key over the columns we are going to return */
     for (i= param->hidden_field_count, reg_field= table->field + i ;
          i < share->fields;
-         i++, reg_field++, m_key_part_info++)
+         i++, reg_field++)
     {
+      if (!((*reg_field)->flags & FIELD_PART_OF_TMP_UNIQUE))
+        continue;
       m_key_part_info->field= *reg_field;
       (*reg_field)->flags |= PART_KEY_FLAG;
       if (m_key_part_info == keyinfo->key_part)
@@ -18904,6 +18960,8 @@ bool Create_tmp_table::finalize(THD *thd,
 	 (ha_base_keytype) m_key_part_info->type == HA_KEYTYPE_VARTEXT1 ||
 	 (ha_base_keytype) m_key_part_info->type == HA_KEYTYPE_VARTEXT2) ?
 	0 : FIELDFLAG_BINARY;
+
+      m_key_part_info++;
     }
   }
 
@@ -27481,7 +27539,7 @@ void st_select_lex::print(THD *thd, String *str, enum_query_type query_type)
     else
       str->append(',');
 
-    if (is_subquery_function() && item->is_autogenerated_name)
+    if (is_subquery_function() && item->is_autogenerated_name())
     {
       /*
         Do not print auto-generated aliases in subqueries. It has no purpose
diff --git a/sql/sql_union.cc b/sql/sql_union.cc
index d5d43e56875..f8025cdf701 100644
--- a/sql/sql_union.cc
+++ b/sql/sql_union.cc
@@ -257,6 +257,8 @@ int select_union_recursive::send_data(List<Item> &values)
       write_err != HA_ERR_FOUND_DUPP_UNIQUE)
   { 
     int err;
+    DBUG_ASSERT(incr_table->s->reclength == table->s->reclength ||
+               incr_table->s->reclength == table->s->reclength - MARIA_UNIQUE_HASH_LENGTH);
     if ((err= incr_table->file->ha_write_tmp_row(table->record[0])))
     {
       bool is_duplicate;
@@ -1561,7 +1563,7 @@ bool st_select_lex_unit::prepare(TABLE_LIST *derived_arg,
     {
       if (with_element)
       {
-        if (with_element->rename_columns_of_derived_unit(thd, this))
+        if (with_element->process_columns_of_derived_unit(thd, this))
           goto err;
         if (check_duplicate_names(thd, sl->item_list, 0))
           goto err;
diff --git a/sql/sql_view.cc b/sql/sql_view.cc
index f47e910c0fe..88a65dcaedb 100644
--- a/sql/sql_view.cc
+++ b/sql/sql_view.cc
@@ -138,7 +138,7 @@ bool check_duplicate_names(THD *thd, List<Item> &item_list, bool gen_unique_view
     Item *check;
     /* treat underlying fields like set by user names */
     if (item->real_item()->type() == Item::FIELD_ITEM)
-      item->is_autogenerated_name= FALSE;
+      item->common_flags&= ~IS_AUTO_GENERATED_NAME;
     itc.rewind();
     while ((check= itc++) && check != item)
     {
@@ -146,9 +146,9 @@ bool check_duplicate_names(THD *thd, List<Item> &item_list, bool gen_unique_view
       {
         if (!gen_unique_view_name)
           goto err;
-        if (item->is_autogenerated_name)
+        if (item->is_autogenerated_name())
           make_unique_view_field_name(thd, item, item_list, item);
-        else if (check->is_autogenerated_name)
+        else if (check->is_autogenerated_name())
           make_unique_view_field_name(thd, check, item_list, item);
         else
           goto err;
@@ -180,7 +180,7 @@ void make_valid_column_names(THD *thd, List<Item> &item_list)
 
   for (uint column_no= 1; (item= it++); column_no++)
   {
-    if (!item->is_autogenerated_name || !check_column_name(item->name.str))
+    if (!item->is_autogenerated_name() || !check_column_name(item->name.str))
       continue;
     name_len= my_snprintf(buff, NAME_LEN, "Name_exp_%u", column_no);
     item->orig_name= item->name.str;
@@ -566,7 +566,7 @@ bool mysql_create_view(THD *thd, TABLE_LIST *views,
     while ((item= it++, name= nm++))
     {
       item->set_name(thd, *name);
-      item->is_autogenerated_name= FALSE;
+      item->common_flags&= ~IS_AUTO_GENERATED_NAME;
     }
   }
 
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 13889ead19f..41891696f22 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -1749,6 +1749,7 @@ End SQL_MODE_ORACLE_SPECIFIC */
         comma_separated_ident_list
         opt_with_column_list
         with_column_list
+        opt_cycle
 
 %type <vers_range_unit> opt_history_unit
 %type <vers_history_point> history_point
@@ -9076,7 +9077,7 @@ select_item:
               if (unlikely(Lex->sql_command == SQLCOM_CREATE_VIEW &&
                           check_column_name($4.str)))
                 my_yyabort_error((ER_WRONG_COLUMN_NAME, MYF(0), $4.str));
-              $2->is_autogenerated_name= FALSE;
+              $2->common_flags&= ~IS_AUTO_GENERATED_NAME;
               $2->set_name(thd, $4);
             }
             else if (!$2->name.str || $2->name.str == item_empty_name)
@@ -10638,7 +10639,7 @@ udf_expr:
             */
             if ($4.str)
             {
-              $2->is_autogenerated_name= FALSE;
+              $2->common_flags&= ~IS_AUTO_GENERATED_NAME;
               $2->set_name(thd, $4);
             }
             /* 
@@ -14749,7 +14750,7 @@ with_list:
 with_list_element:
 	  query_name
 	  opt_with_column_list 
-          AS '(' query_expression ')'
+          AS '(' query_expression ')' opt_cycle
  	  {
             LEX *lex= thd->lex;
             const char *query_start= lex->sphead ? lex->sphead->m_tmp_query
@@ -14761,9 +14762,30 @@ with_list_element:
             if (elem->set_unparsed_spec(thd, spec_start, $6.pos(),
                                         spec_start - query_start))
               MYSQL_YYABORT;
+            if ($7)
+            {
+              elem->set_cycle_list($7);
+            }
 	  }
 	;
 
+opt_cycle:
+         /* empty */
+         { $$= NULL; }
+         |
+         CYCLE_SYM
+         {
+           if (!Lex->curr_with_clause->with_recursive)
+           {
+             thd->parse_error(ER_SYNTAX_ERROR, $1.pos());
+           }
+         }
+         comma_separated_ident_list RESTRICT
+         {
+           $$= $3;
+         }
+         ;
+
 
 opt_with_column_list:
           /* empty */