Don't cache savepoint identifiers

Fixed bug in compiler where the string identifier of a savepoint would
be cached in the identifier quoting dictionary; as these identifiers
are arbitrary, a small memory leak could occur if a single
:class:`.Connection` had an unbounded number of savepoints used,
as well as if the savepoint clause constructs were used directly
with an unbounded umber of savepoint names.   The memory leak does
**not** impact the vast majority of cases as normally the
:class:`.Connection`, which renders savepoint names with a simple
counter starting at "1", is used on a per-transaction or
per-fixed-number-of-transactions basis before being discarded.

The savepoint name in virtually all cases does not require quoting
at all, however to support potential third party use cases
the "check for quotes needed" logic is retained, at a small
performance cost.   Uncondtionally quoting the name is another
option, but this would turn the name into a case sensitive name
which runs the risk of poor interactions with existing deployments
that may be looking at these names in other contexts.

Change-Id: I6b53c96abf7fdf1840592bbca5da81347911844c
Fixes: #3931

3 files changed, 68 insertions, 5 deletions

diff --git a/doc/build/changelog/changelog_11.rst b/doc/build/changelog/changelog_11.rst
index ffb4bce94..22f4e2bc4 100644
--- a/doc/build/changelog/changelog_11.rst
+++ b/doc/build/changelog/changelog_11.rst
@@ -23,6 +23,21 @@
 
     .. change::
         :tags: bug, sql
+        :tickets: 3931
+
+        Fixed bug in compiler where the string identifier of a savepoint would
+        be cached in the identifier quoting dictionary; as these identifiers
+        are arbitrary, a small memory leak could occur if a single
+        :class:`.Connection` had an unbounded number of savepoints used,
+        as well as if the savepoint clause constructs were used directly
+        with an unbounded umber of savepoint names.   The memory leak does
+        **not** impact the vast majority of cases as normally the
+        :class:`.Connection`, which renders savepoint names with a simple
+        counter starting at "1", is used on a per-transaction or
+        per-fixed-number-of-transactions basis before being discarded.
+
+    .. change::
+        :tags: bug, sql
         :tickets: 3924
 
         Fixed bug in new "schema translate" feature where the translated schema
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index aeb40030f..bfa22c206 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -2933,7 +2933,13 @@ class IdentifierPreparer(object):
         return self.quote(name or alias.name)
 
     def format_savepoint(self, savepoint, name=None):
-        return self.quote(name or savepoint.ident)
+        # Running the savepoint name through quoting is unnecessary
+        # for all known dialects.  This is here to support potential
+        # third party use cases
+        ident = name or savepoint.ident
+        if self._requires_quotes(ident):
+            ident = self.quote_identifier(ident)
+        return ident
 
     @util.dependencies("sqlalchemy.sql.naming")
     def format_constraint(self, naming, constraint):
diff --git a/test/aaa_profiling/test_memusage.py b/test/aaa_profiling/test_memusage.py
index 7013159dd..53f118e15 100644
--- a/test/aaa_profiling/test_memusage.py
+++ b/test/aaa_profiling/test_memusage.py
@@ -34,7 +34,8 @@ class ASub(A):
     pass
 
 
-def profile_memory(maxtimes=50):
+def profile_memory(maxtimes=50,
+                   assert_no_sessions=True, get_num_objects=None):
     def decorate(func):
         # run the test N times.  if length of gc.get_objects()
         # keeps growing, assert false
@@ -56,15 +57,19 @@ def profile_memory(maxtimes=50):
             samples = []
 
             success = False
-            for y in range(maxtimes // 5):
+            for y in range(100 // 5):
                 for x in range(5):
                     func(*args)
                     gc_collect()
-                    samples.append(len(get_objects_skipping_sqlite_issue()))
+                    samples.append(
+                        get_num_objects() if get_num_objects is not None
+                        else len(get_objects_skipping_sqlite_issue())
+                    )
 
                 print("sample gc sizes:", samples)
 
-                assert len(_sessions) == 0
+                if assert_no_sessions:
+                    assert len(_sessions) == 0
 
                 # check for "flatline" - size is constant for
                 # 5 iterations
@@ -341,6 +346,43 @@ class MemUsageTest(EnsureZeroed):
         finally:
             metadata.drop_all()
 
+    @testing.requires.savepoints
+    @testing.provide_metadata
+    def test_savepoints(self):
+        metadata = self.metadata
+
+        some_table = Table(
+            't', metadata,
+            Column('id', Integer, primary_key=True,
+                   test_needs_autoincrement=True)
+        )
+
+        class SomeClass(object):
+            pass
+
+        mapper(SomeClass, some_table)
+
+        metadata.create_all()
+
+        session = Session(testing.db)
+
+        target_strings = session.connection().\
+            dialect.identifier_preparer._strings
+
+        with session.transaction:
+            @profile_memory(
+                assert_no_sessions=False,
+                get_num_objects=lambda: len(target_strings))
+            def go():
+
+                sc = SomeClass()
+                session.add(sc)
+
+                with session.begin_nested():
+                    session.query(SomeClass).first()
+
+            go()
+
     @testing.crashes('mysql+cymysql', 'blocking')
     def test_unicode_warnings(self):
         metadata = MetaData(self.engine)