Track if we're rendering within the CTE recursively

Fixed a regression that occurred from the previous fix to :ticket:`4204` in
version 1.2.5, where a CTE that refers to itself after the
:meth:`.CTE.alias` method has been called would not refer to iself
correctly.

Change-Id: Iaa63d65ad2b90c8693f9953fbb32dbb10c73a037
Fixes: #4204

4 files changed, 63 insertions, 2 deletions

diff --git a/doc/build/changelog/unreleased_12/4204.rst b/doc/build/changelog/unreleased_12/4204.rst
new file mode 100644
index 000000000..975330b3a
--- /dev/null
+++ b/doc/build/changelog/unreleased_12/4204.rst
@@ -0,0 +1,9 @@
+.. change::
+    :tags: bug, sql
+    :tickets: 4204
+    :versions: 1.3.0b1
+
+    Fixed a regression that occurred from the previous fix to :ticket:`4204` in
+    version 1.2.5, where a CTE that refers to itself after the
+    :meth:`.CTE.alias` method has been called would not refer to iself
+    correctly.
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index 438484f33..6c7e6145d 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -1346,18 +1346,23 @@ class SQLCompiler(Compiled):
             return self.bindtemplate % {'name': name}
 
     def visit_cte(self, cte, asfrom=False, ashint=False,
-                  fromhints=None,
+                  fromhints=None, visiting_cte=None,
                   **kwargs):
         self._init_cte_state()
 
+        kwargs['visiting_cte'] = cte
         if isinstance(cte.name, elements._truncated_label):
             cte_name = self._truncated_identifier("alias", cte.name)
         else:
             cte_name = cte.name
 
         is_new_cte = True
+        embedded_in_current_named_cte = False
+
         if cte_name in self.ctes_by_name:
             existing_cte = self.ctes_by_name[cte_name]
+            embedded_in_current_named_cte = visiting_cte is existing_cte
+
             # we've generated a same-named CTE that we are enclosed in,
             # or this is the same CTE.  just return the name.
             if cte in existing_cte._restates or cte is existing_cte:
@@ -1431,6 +1436,9 @@ class SQLCompiler(Compiled):
                 self.ctes[cte] = text
 
         if asfrom:
+            if not is_new_cte and embedded_in_current_named_cte:
+                return self.preparer.format_alias(cte, cte_name)
+
             if cte_pre_alias_name:
                 text = self.preparer.format_alias(cte, cte_pre_alias_name)
                 if self.preparer._requires_quotes(cte_name):
diff --git a/test/sql/test_cte.py b/test/sql/test_cte.py
index 2c19ed032..efc4640ed 100644
--- a/test/sql/test_cte.py
+++ b/test/sql/test_cte.py
@@ -122,6 +122,50 @@ class CTETest(fixtures.TestBase, AssertsCompiledSQL):
             "JOIN parts ON anon_1.part = parts.part "
             "GROUP BY anon_1.sub_part", dialect=mssql.dialect())
 
+    def test_recursive_inner_cte_unioned_to_alias(self):
+        parts = table('parts',
+                      column('part'),
+                      column('sub_part'),
+                      column('quantity'),
+                      )
+
+        included_parts = select([
+            parts.c.sub_part,
+            parts.c.part,
+            parts.c.quantity]).\
+            where(parts.c.part == 'our part').\
+            cte(recursive=True)
+
+        incl_alias = included_parts.alias('incl')
+        parts_alias = parts.alias()
+        included_parts = incl_alias.union(
+            select([
+                parts_alias.c.sub_part,
+                parts_alias.c.part,
+                parts_alias.c.quantity]).
+            where(parts_alias.c.part == incl_alias.c.sub_part)
+        )
+
+        s = select([
+            included_parts.c.sub_part,
+            func.sum(included_parts.c.quantity).label('total_quantity')]).\
+            select_from(included_parts.join(
+                parts, included_parts.c.part == parts.c.part)).\
+            group_by(included_parts.c.sub_part)
+        self.assert_compile(
+            s, "WITH RECURSIVE incl(sub_part, part, quantity) "
+            "AS (SELECT parts.sub_part AS sub_part, parts.part "
+            "AS part, parts.quantity AS quantity FROM parts "
+            "WHERE parts.part = :part_1 UNION "
+            "SELECT parts_1.sub_part AS sub_part, "
+            "parts_1.part AS part, parts_1.quantity "
+            "AS quantity FROM parts AS parts_1, incl "
+            "WHERE parts_1.part = incl.sub_part) "
+            "SELECT incl.sub_part, "
+            "sum(incl.quantity) AS total_quantity FROM incl "
+            "JOIN parts ON incl.part = parts.part "
+            "GROUP BY incl.sub_part")
+
     def test_recursive_union_no_alias_one(self):
         s1 = select([literal(0).label("x")])
         cte = s1.cte(name="cte", recursive=True)
diff --git a/test/sql/test_generative.py b/test/sql/test_generative.py
index 145b2da3c..8b1436879 100644
--- a/test/sql/test_generative.py
+++ b/test/sql/test_generative.py
@@ -509,7 +509,7 @@ class ClauseTest(fixtures.TestBase, AssertsCompiledSQL):
         self.assert_compile(
             cloned,
             "WITH RECURSIVE foo(n) AS (SELECT values(:values_1) AS n "
-            "UNION ALL SELECT foo.n + :n_1 AS anon_1 FROM t AS foo "
+            "UNION ALL SELECT foo.n + :n_1 AS anon_1 FROM foo "
             "WHERE foo.n < :n_2) SELECT sum(foo.n) AS sum_1 FROM foo"
         )