Reverse Alias nesting concept

The Alias object no longer has "element" and "original", it now
has "wrapped" and "element" (the name .original is also left
as a descriptor for legacy access by third party dialects).
These two data members refer to the
dual roles Alias needs to play, where in the Python sense it needs
to refer to the thing it was applied against directly, whereas in the
SQL sense it needs to refer to the ultimate "non-alias" thing it
refers towards.   Both are necessary to maintain.  However, the change
here has each Alias object access the non-Alias object immediately
so that the "unwrapping" is simpler and does not need any special
logic.

In the SQL sense, Alias objects don't nest, the only potential
was that of the CTE, however there is no such thing as
a nested CTE, see link below.

This change is an interim change along the way to breaking Alias
into more classes and breaking away Select objects from being
FromClause objects.

Change-Id: Ie7a0d064226cb074ca745505129b5ec7d879e389
References: https://stackoverflow.com/questions/1413516/can-you-create-nested-with-clauses-for-common-table-expressions

6 files changed, 114 insertions, 33 deletions

diff --git a/lib/sqlalchemy/dialects/firebird/base.py b/lib/sqlalchemy/dialects/firebird/base.py
index 7ae198f6c..e67bb2d38 100644
--- a/lib/sqlalchemy/dialects/firebird/base.py
+++ b/lib/sqlalchemy/dialects/firebird/base.py
@@ -469,12 +469,12 @@ class FBCompiler(sql.compiler.SQLCompiler):
                 )
 
                 return (
-                    self.process(alias.original, asfrom=asfrom, **kwargs)
+                    self.process(alias.element, asfrom=asfrom, **kwargs)
                     + " "
                     + self.preparer.format_alias(alias, alias_name)
                 )
             else:
-                return self.process(alias.original, **kwargs)
+                return self.process(alias.element, **kwargs)
 
     def visit_substring_func(self, func, **kw):
         s = self.process(func.clauses.clauses[0])
diff --git a/lib/sqlalchemy/dialects/mssql/base.py b/lib/sqlalchemy/dialects/mssql/base.py
index 00a110aa2..8c3d5b3fa 100644
--- a/lib/sqlalchemy/dialects/mssql/base.py
+++ b/lib/sqlalchemy/dialects/mssql/base.py
@@ -1659,7 +1659,7 @@ class MSSQLCompiler(compiler.SQLCompiler):
     @_with_legacy_schema_aliasing
     def visit_alias(self, alias, **kw):
         # translate for schema-qualified table aliases
-        kw["mssql_aliased"] = alias.original
+        kw["mssql_aliased"] = alias.element
         return super(MSSQLCompiler, self).visit_alias(alias, **kw)
 
     @_with_legacy_schema_aliasing
diff --git a/lib/sqlalchemy/sql/coercions.py b/lib/sqlalchemy/sql/coercions.py
index d4551eb60..ccd506a73 100644
--- a/lib/sqlalchemy/sql/coercions.py
+++ b/lib/sqlalchemy/sql/coercions.py
@@ -176,10 +176,10 @@ class _ColumnCoercions(object):
         elif (
             resolved._is_from_clause
             and isinstance(resolved, selectable.Alias)
-            and resolved.original._is_select_statement
+            and resolved.element._is_select_statement
         ):
             self._warn_for_scalar_subquery_coercion()
-            return resolved.original.scalar_subquery()
+            return resolved.element.scalar_subquery()
         else:
             self._raise_for_expected(original_element, argname)
 
@@ -282,9 +282,9 @@ class InElementImpl(RoleImpl, roles.InElementRole):
         if resolved._is_from_clause:
             if (
                 isinstance(resolved, selectable.Alias)
-                and resolved.original._is_select_statement
+                and resolved.element._is_select_statement
             ):
-                return resolved.original
+                return resolved.element
             else:
                 return resolved.select()
         else:
@@ -579,9 +579,9 @@ class DMLSelectImpl(_NoTextCoercion, RoleImpl, roles.DMLSelectRole):
         if resolved._is_from_clause:
             if (
                 isinstance(resolved, selectable.Alias)
-                and resolved.original._is_select_statement
+                and resolved.element._is_select_statement
             ):
-                return resolved.original
+                return resolved.element
             else:
                 return resolved.select()
         else:
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index 8080d2cc6..6fcf1a524 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -1680,7 +1680,7 @@ class SQLCompiler(Compiled):
                 if self.positional:
                     kwargs["positional_names"] = self.cte_positional[cte] = []
 
-                text += " AS \n" + cte.original._compiler_dispatch(
+                text += " AS \n" + cte.element._compiler_dispatch(
                     self, asfrom=True, **kwargs
                 )
 
@@ -1722,7 +1722,7 @@ class SQLCompiler(Compiled):
         if ashint:
             return self.preparer.format_alias(alias, alias_name)
         elif asfrom:
-            ret = alias.original._compiler_dispatch(
+            ret = alias.element._compiler_dispatch(
                 self, asfrom=True, **kwargs
             ) + self.get_render_as_alias_suffix(
                 self.preparer.format_alias(alias, alias_name)
@@ -1735,7 +1735,7 @@ class SQLCompiler(Compiled):
 
             return ret
         else:
-            return alias.original._compiler_dispatch(self, **kwargs)
+            return alias.element._compiler_dispatch(self, **kwargs)
 
     def visit_lateral(self, lateral, **kw):
         kw["lateral"] = True
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index b0d6002b7..014c782d0 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -1261,26 +1261,30 @@ class Alias(roles.AnonymizedFromClauseRole, FromClause):
         )
 
     def _init(self, selectable, name=None):
-        baseselectable = selectable
-        while isinstance(baseselectable, Alias):
-            baseselectable = baseselectable.element
-        self.original = baseselectable
-        self.supports_execution = baseselectable.supports_execution
+        self.wrapped = selectable
+        if isinstance(selectable, Alias):
+            selectable = selectable.element
+            assert not isinstance(selectable, Alias)
+
+        self.supports_execution = selectable.supports_execution
         if self.supports_execution:
-            self._execution_options = baseselectable._execution_options
+            self._execution_options = selectable._execution_options
         self.element = selectable
         self._orig_name = name
         if name is None:
-            if self.original.named_with_column:
-                name = getattr(self.original, "name", None)
+            if (
+                isinstance(selectable, FromClause)
+                and selectable.named_with_column
+            ):
+                name = getattr(selectable, "name", None)
             name = _anonymous_label("%%(%d %s)s" % (id(self), name or "anon"))
         self.name = name
 
     def self_group(self, against=None):
         if (
             isinstance(against, CompoundSelect)
-            and isinstance(self.original, Select)
-            and self.original._needs_parens_for_grouping()
+            and isinstance(self.element, Select)
+            and self.element._needs_parens_for_grouping()
         ):
             return FromGrouping(self)
 
@@ -1293,17 +1297,22 @@ class Alias(roles.AnonymizedFromClauseRole, FromClause):
         else:
             return self.name.encode("ascii", "backslashreplace")
 
+    @property
+    def original(self):
+        """legacy for dialects that are referring to Alias.original"""
+        return self.element
+
     def is_derived_from(self, fromclause):
         if fromclause in self._cloned_set:
             return True
         return self.element.is_derived_from(fromclause)
 
     def _populate_column_collection(self):
-        for col in self.element.columns._all_columns:
+        for col in self.wrapped.columns._all_columns:
             col._make_proxy(self)
 
     def _refresh_for_new_column(self, column):
-        col = self.element._refresh_for_new_column(column)
+        col = self.wrapped._refresh_for_new_column(column)
         if col is not None:
             if not self._cols_populated:
                 return None
@@ -1319,17 +1328,17 @@ class Alias(roles.AnonymizedFromClauseRole, FromClause):
         if isinstance(self.element, TableClause):
             return
         self._reset_exported()
-        self.element = clone(self.element, **kw)
-        baseselectable = self.element
-        while isinstance(baseselectable, Alias):
-            baseselectable = baseselectable.element
-        self.original = baseselectable
+        self.wrapped = clone(self.wrapped, **kw)
+        if isinstance(self.wrapped, Alias):
+            self.element = self.wrapped.element
+        else:
+            self.element = self.wrapped
 
     def get_children(self, column_collections=True, **kw):
         if column_collections:
             for c in self.c:
                 yield c
-        yield self.element
+        yield self.wrapped
 
     def _cache_key(self, **kw):
         return (self.__class__, self.element._cache_key(**kw), self._orig_name)
@@ -1522,7 +1531,7 @@ class CTE(Generative, HasSuffixes, Alias):
 
     def alias(self, name=None, flat=False):
         return CTE._construct(
-            self.original,
+            self.element,
             name=name,
             recursive=self.recursive,
             _cte_alias=self,
@@ -1531,7 +1540,7 @@ class CTE(Generative, HasSuffixes, Alias):
 
     def union(self, other):
         return CTE._construct(
-            self.original.union(other),
+            self.element.union(other),
             name=self.name,
             recursive=self.recursive,
             _restates=self._restates.union([self]),
@@ -1540,7 +1549,7 @@ class CTE(Generative, HasSuffixes, Alias):
 
     def union_all(self, other):
         return CTE._construct(
-            self.original.union_all(other),
+            self.element.union_all(other),
             name=self.name,
             recursive=self.recursive,
             _restates=self._restates.union([self]),
diff --git a/test/sql/test_selectable.py b/test/sql/test_selectable.py
index 186fb3d9e..f525703f1 100644
--- a/test/sql/test_selectable.py
+++ b/test/sql/test_selectable.py
@@ -931,6 +931,22 @@ class RefreshForNewColTest(fixtures.TestBase):
         s._refresh_for_new_column(q)
         assert q in s.c.b_x.proxy_set
 
+    def test_alias_alias_samename_init(self):
+        a = table("a", column("x"))
+        b = table("b", column("y"))
+        s1 = select([a, b]).apply_labels().alias()
+        s2 = s1.alias()
+
+        s1.c
+        s2.c
+
+        q = column("x")
+        b.append_column(q)
+
+        s2._refresh_for_new_column(q)
+
+        is_(s1.corresponding_column(s2.c.b_x), s1.c.b_x)
+
     def test_aliased_select_samename_uninit(self):
         a = table("a", column("x"))
         b = table("b", column("y"))
@@ -2584,3 +2600,59 @@ class ForUpdateTest(fixtures.TestBase, AssertsCompiledSQL):
             "SELECT t_1.c FROM t AS t_1 FOR SHARE OF t_1",
             dialect="postgresql",
         )
+
+
+class AliasTest(fixtures.TestBase, AssertsCompiledSQL):
+    __dialect__ = "default"
+
+    def test_legacy_original_accessor(self):
+        t = table("t", column("c"))
+        a1 = t.alias()
+        a2 = a1.alias()
+        a3 = a2.alias()
+
+        is_(a1.original, t)
+        is_(a2.original, t)
+        is_(a3.original, t)
+
+    def test_wrapped(self):
+        t = table("t", column("c"))
+        a1 = t.alias()
+        a2 = a1.alias()
+        a3 = a2.alias()
+
+        is_(a1.element, t)
+        is_(a2.element, t)
+        is_(a3.element, t)
+
+        is_(a3.wrapped, a2)
+        is_(a2.wrapped, a1)
+        is_(a1.wrapped, t)
+
+    def test_get_children_preserves_wrapped(self):
+        t = table("t", column("c"))
+        stmt = select([t])
+        a1 = stmt.alias()
+        a2 = a1.alias()
+        eq_(set(a2.get_children(column_collections=False)), {a1})
+
+    def test_wrapped_correspondence(self):
+        t = table("t", column("c"))
+        stmt = select([t])
+        a1 = stmt.alias()
+        a2 = a1.alias()
+
+        is_(a1.corresponding_column(a2.c.c), a1.c.c)
+
+    def test_copy_internals_preserves_wrapped(self):
+        t = table("t", column("c"))
+        stmt = select([t])
+        a1 = stmt.alias()
+        a2 = a1.alias()
+
+        is_(a2.element, a2.wrapped.element)
+
+        a3 = a2._clone()
+        a3._copy_internals()
+        is_(a1.corresponding_column(a3.c.c), a1.c.c)
+        is_(a3.element, a3.wrapped.element)