From: Mike Bayer <mike_mp@zzzcomputing.com>
Date: Tue, 4 Mar 2008 18:20:09 +0000 (+0000)
Subject: - fixed bug which was preventing UNIONS from being cloneable,
X-Git-Tag: rel_0_4_4~40
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=8e0ea84c333aac2b28326084a03c48aa4a212f45;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git

- fixed bug which was preventing UNIONS from being cloneable,
[ticket:986]
---

diff --git a/CHANGES b/CHANGES
index de8c168eaa..e53812101d 100644
--- a/CHANGES
+++ b/CHANGES
@@ -19,6 +19,9 @@ CHANGES
     - implemented two-phase API for "threadlocal" engine, 
       via engine.begin_twophase(), engine.prepare()
       [ticket:936]
+
+    - fixed bug which was preventing UNIONS from being cloneable,
+      [ticket:986]
       
 - orm
     - any(), has(), contains(), attribute level == and != now
diff --git a/lib/sqlalchemy/sql/expression.py b/lib/sqlalchemy/sql/expression.py
index 56629a6ca3..812c70c2d8 100644
--- a/lib/sqlalchemy/sql/expression.py
+++ b/lib/sqlalchemy/sql/expression.py
@@ -2925,8 +2925,6 @@ class CompoundSelect(_SelectBaseMixin, FromClause):
             else:
                 self.selects.append(s)
 
-        self._col_map = {}
-
         _SelectBaseMixin.__init__(self, **kwargs)
 
         for s in self.selects:
@@ -2942,11 +2940,15 @@ class CompoundSelect(_SelectBaseMixin, FromClause):
                 yield c
 
     def _proxy_column(self, column):
-        selectable = column.table
-        col_ordering = self._col_map.get(selectable, None)
-        if col_ordering is None:
-            self._col_map[selectable] = col_ordering = []
-
+        if not hasattr(self, '_col_map'):
+            self._col_map = dict([(s, []) for s in self.selects])
+            for s in self.selects:
+                for c in s.c + [s.oid_column]:
+                    self._col_map[c] = s
+        
+        selectable = self._col_map[column]
+        col_ordering = self._col_map[selectable]
+        
         if selectable is self.selects[0]:
             if self.use_labels:
                 col = column._make_proxy(self, name=column._label)
@@ -2962,8 +2964,9 @@ class CompoundSelect(_SelectBaseMixin, FromClause):
 
     def _copy_internals(self, clone=_clone):
         self._clone_from_clause()
-        self._col_map = {}
         self.selects = [clone(s) for s in self.selects]
+        if hasattr(self, '_col_map'):
+            del self._col_map
         for attr in ('_order_by_clause', '_group_by_clause'):
             if getattr(self, attr) is not None:
                 setattr(self, attr, clone(getattr(self, attr)))
diff --git a/lib/sqlalchemy/sql/util.py b/lib/sqlalchemy/sql/util.py
index 70a1dcc964..9954811d64 100644
--- a/lib/sqlalchemy/sql/util.py
+++ b/lib/sqlalchemy/sql/util.py
@@ -152,6 +152,12 @@ class AbstractClauseProcessor(object):
 
     This class implements its own visit-and-copy strategy but maintains the
     same public interface as visitors.ClauseVisitor.
+    
+    The convert_element() method receives the *un-copied* version of each element.
+    It can return a new element or None for no change.  If None, the element
+    will be cloned afterwards and added to the new structure.  Note this is the
+    opposite behavior of visitors.traverse(clone=True), where visitors receive
+    the cloned element so that it can be mutated.
     """
 
     __traverse_options__ = {'column_collections':False}
diff --git a/test/sql/generative.py b/test/sql/generative.py
index 831c2e2873..5e6b3b7e6c 100644
--- a/test/sql/generative.py
+++ b/test/sql/generative.py
@@ -224,7 +224,29 @@ class ClauseTest(TestBase, AssertsCompiledSQL):
         print str(s5)
         assert str(s5) == s5_assert
         assert str(s4) == s4_assert
-
+    
+    def test_union(self):
+        u = union(t1.select(), t2.select())
+        u2 = ClauseVisitor().traverse(u, clone=True)
+        assert str(u) == str(u2)
+        assert [str(c) for c in u2.c] == [str(c) for c in u.c]
+
+        u = union(t1.select(), t2.select())
+        cols = [str(c) for c in u.c]
+        u2 = ClauseVisitor().traverse(u, clone=True)
+        assert str(u) == str(u2)
+        assert [str(c) for c in u2.c] == cols
+        
+        s1 = select([t1], t1.c.col1 == bindparam('id_param'))
+        s2 = select([t2])
+        u = union(s1, s2)
+        
+        u2 = u.params(id_param=7)
+        u3 = u.params(id_param=10)
+        assert str(u) == str(u2) == str(u3)
+        assert u2.compile().params == {'id_param':7}
+        assert u3.compile().params == {'id_param':10}
+        
     def test_binds(self):
         """test that unique bindparams change their name upon clone() to prevent conflicts"""