Ensure CTE internals are handled during clone

The CTE construct was missing a _copy_internals() method
which would handle CTE-specific structures including _cte_alias,
_restates during a clone operation.

Change-Id: I9aeac9cd24d8f7ae6b70e52650d61f7c96cb6d7e
Fixes: #3722

3 files changed, 34 insertions, 0 deletions

diff --git a/doc/build/changelog/changelog_10.rst b/doc/build/changelog/changelog_10.rst
index b59ab392e..87ca1cb31 100644
--- a/doc/build/changelog/changelog_10.rst
+++ b/doc/build/changelog/changelog_10.rst
@@ -20,6 +20,15 @@
 
     .. change::
         :tags: bug, sql
+        :tickets: 3722
+
+        Fixed bug in :class:`.CTE` structure which would cause it to not
+        clone properly when a union was used, as is common in a recursive
+        CTE.  The improper cloning would cause errors when the CTE is used
+        in various ORM contexts such as that of a :func:`.column_property`.
+
+    .. change::
+        :tags: bug, sql
         :tickets: 3721
 
         Fixed bug whereby :meth:`.Table.tometadata` would make a duplicate
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index 6ef327b95..f75613e35 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -1269,6 +1269,14 @@ class CTE(Generative, HasSuffixes, Alias):
             self._suffixes = _suffixes
         super(CTE, self).__init__(selectable, name=name)
 
+    def _copy_internals(self, clone=_clone, **kw):
+        super(CTE, self)._copy_internals(clone, **kw)
+        if self._cte_alias is not None:
+            self._cte_alias = self
+        self._restates = frozenset([
+            clone(elem, **kw) for elem in self._restates
+        ])
+
     @util.dependencies("sqlalchemy.sql.dml")
     def _populate_column_collection(self, dml):
         if isinstance(self.element, dml.UpdateBase):
diff --git a/test/sql/test_generative.py b/test/sql/test_generative.py
index 9cf1ef612..81c589d11 100644
--- a/test/sql/test_generative.py
+++ b/test/sql/test_generative.py
@@ -475,6 +475,23 @@ class ClauseTest(fixtures.TestBase, AssertsCompiledSQL):
             "FROM table3 AS table3_1"
         )
 
+    def test_cte_w_union(self):
+        t = select([func.values(1).label("n")]).cte("t", recursive=True)
+        t = t.union_all(select([t.c.n + 1]).where(t.c.n < 100))
+        s = select([func.sum(t.c.n)])
+
+        from sqlalchemy.sql.visitors import cloned_traverse
+        cloned = cloned_traverse(s, {}, {})
+
+        self.assert_compile(cloned,
+                            "WITH RECURSIVE t(n) AS "
+                            "(SELECT values(:values_1) AS n "
+                            "UNION ALL SELECT t.n + :n_1 AS anon_1 "
+                            "FROM t "
+                            "WHERE t.n < :n_2) "
+                            "SELECT sum(t.n) AS sum_1 FROM t"
+                            )
+
     def test_text(self):
         clause = text(
             "select * from table where foo=:bar",