Improve serializer behavior

Fix an issue where the Annotated system needs to have a
__reduce__ method, also see why we can't default to HIGHEST_PROTOCOL.
This latter part might not be a good idea until 1.2 for compatibility
reasons.

Change-Id: I0239e38259fc768c9e3b6c448c29161e271a969c
Fixes: #3918

4 files changed, 45 insertions, 17 deletions

diff --git a/doc/build/changelog/changelog_12.rst b/doc/build/changelog/changelog_12.rst
index b99e9ef80..1ed24a1e2 100644
--- a/doc/build/changelog/changelog_12.rst
+++ b/doc/build/changelog/changelog_12.rst
@@ -13,6 +13,15 @@
 .. changelog::
     :version: 1.2.0b1
 
+    .. change:: 3918
+        :tags: bug, ext
+        :tickets: 3918
+
+        Fixed a bug in the ``sqlalchemy.ext.serializer`` extension whereby
+        an "annotated" SQL element (as produced by the ORM for many types
+        of SQL expressions) could not be reliably serialized.  Also bumped
+        the default pickle level for the serializer to "HIGHEST_PROTOCOL".
+
     .. change:: 3932
         :tags: bug, oracle
         :tickets: 3932
diff --git a/lib/sqlalchemy/ext/serializer.py b/lib/sqlalchemy/ext/serializer.py
index 2fbc62ec6..ed8fef85c 100644
--- a/lib/sqlalchemy/ext/serializer.py
+++ b/lib/sqlalchemy/ext/serializer.py
@@ -146,7 +146,7 @@ def Deserializer(file, metadata=None, scoped_session=None, engine=None):
     return unpickler
 
 
-def dumps(obj, protocol=0):
+def dumps(obj, protocol=pickle.HIGHEST_PROTOCOL):
     buf = byte_buffer()
     pickler = Serializer(buf, protocol)
     pickler.dump(obj)
diff --git a/lib/sqlalchemy/sql/annotation.py b/lib/sqlalchemy/sql/annotation.py
index e6f6311c4..6d0eaa1d2 100644
--- a/lib/sqlalchemy/sql/annotation.py
+++ b/lib/sqlalchemy/sql/annotation.py
@@ -94,6 +94,9 @@ class Annotated(object):
             clone.__dict__.update(self.__dict__)
             return self.__class__(clone, self._annotations)
 
+    def __reduce__(self):
+        return self.__class__, (self.__element, self._annotations)
+
     def __hash__(self):
         return self._hash
 
diff --git a/test/ext/test_serializer.py b/test/ext/test_serializer.py
index 4e33473b8..1ea5dfd1d 100644
--- a/test/ext/test_serializer.py
+++ b/test/ext/test_serializer.py
@@ -3,17 +3,21 @@
 from sqlalchemy.ext import serializer
 from sqlalchemy import testing
 from sqlalchemy import Integer, String, ForeignKey, select, \
-    desc, func, util, MetaData, literal_column
+    desc, func, util, MetaData, literal_column, join
 from sqlalchemy.testing.schema import Table
 from sqlalchemy.testing.schema import Column
 from sqlalchemy.orm import relationship, sessionmaker, scoped_session, \
     class_mapper, mapper, joinedload, configure_mappers, aliased
 from sqlalchemy.testing import eq_, AssertsCompiledSQL
 from sqlalchemy.util import u, ue
-
 from sqlalchemy.testing import fixtures
 
 
+def pickle_protocols():
+    return iter([-1, 1, 2])
+    #return iter([-1, 0, 1, 2])
+
+
 class User(fixtures.ComparableEntity):
     pass
 
@@ -127,20 +131,32 @@ class SerializeTest(AssertsCompiledSQL, fixtures.MappedTest):
         eq_(q2.all(), [User(name='fred')])
         eq_(list(q2.values(User.id, User.name)), [(9, 'fred')])
 
-    # fails too often/randomly
-    # @testing.requires.non_broken_pickle
-    # def test_query_three(self):
-    #    ua = aliased(User)
-    #    q = \
-    #        Session.query(ua).join(ua.addresses).\
-    #           filter(Address.email.like('%fred%'))
-    #    q2 = serializer.loads(serializer.dumps(q, -1), users.metadata,
-    #                          Session)
-    #    eq_(q2.all(), [User(name='fred')])
-    #
-        # try to pull out the aliased entity here...
-    #    ua_2 = q2._entities[0].entity_zero.entity
-    #    eq_(list(q2.values(ua_2.id, ua_2.name)), [(9, 'fred')])
+    @testing.requires.non_broken_pickle
+    def test_query_three(self):
+        ua = aliased(User)
+        q = \
+            Session.query(ua).join(ua.addresses).\
+               filter(Address.email.like('%fred%'))
+        for prot in pickle_protocols():
+            q2 = serializer.loads(serializer.dumps(q, prot), users.metadata,
+                                  Session)
+            eq_(q2.all(), [User(name='fred')])
+
+           # try to pull out the aliased entity here...
+            ua_2 = q2._entities[0].entity_zero.entity
+            eq_(list(q2.values(ua_2.id, ua_2.name)), [(9, 'fred')])
+
+    def test_annotated_one(self):
+        j = join(users, addresses)._annotate({"foo": "bar"})
+        query = select([addresses]).select_from(
+            j
+        )
+
+        str(query)
+        for prot in pickle_protocols():
+            pickled_failing = serializer.dumps(
+                j, prot)
+            serializer.loads(pickled_failing, users.metadata, None)
 
     @testing.requires.non_broken_pickle
     def test_orm_join(self):