Convert schema_translate to a post compile

Revised the :paramref:`.Connection.execution_options.schema_translate_map`
feature such that the processing of the SQL statement to receive a specific
schema name occurs within the execution phase of the statement, rather than
at the compile phase.   This is to support the statement being efficiently
cached.   Previously, the current schema being rendered into the statement
for a particular run would be considered as part of the cache key itself,
meaning that for a run against hundreds of schemas, there would be hundreds
of cache keys, rendering the cache much less performant.  The new behavior
is that the rendering is done in a similar  manner as the "post compile"
rendering added in 1.4 as part of :ticket:`4645`, :ticket:`4808`.

Fixes: #5004
Change-Id: Ia5c89eb27cc8dc2c5b8e76d6c07c46290a7901b6

4 files changed, 48 insertions, 54 deletions

diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py
index aa21fb13b..4ed3b9af7 100644
--- a/lib/sqlalchemy/engine/base.py
+++ b/lib/sqlalchemy/engine/base.py
@@ -17,7 +17,6 @@ from .. import inspection
 from .. import log
 from .. import util
 from ..sql import compiler
-from ..sql import schema
 from ..sql import util as sql_util
 
 
@@ -51,21 +50,7 @@ class Connection(Connectable):
 
     """
 
-    schema_for_object = schema._schema_getter(None)
-    """Return the ".schema" attribute for an object.
-
-    Used for :class:`.Table`, :class:`.Sequence` and similar objects,
-    and takes into account
-    the :paramref:`.Connection.execution_options.schema_translate_map`
-    parameter.
-
-      .. versionadded:: 1.1
-
-      .. seealso::
-
-          :ref:`schema_translating`
-
-    """
+    _schema_translate_map = None
 
     def __init__(
         self,
@@ -92,7 +77,7 @@ class Connection(Connectable):
             self.should_close_with_result = False
             self.dispatch = _dispatch
             self._has_events = _branch_from._has_events
-            self.schema_for_object = _branch_from.schema_for_object
+            self._schema_translate_map = _branch_from._schema_translate_map
         else:
             self.__connection = (
                 connection
@@ -122,6 +107,24 @@ class Connection(Connectable):
         if self._has_events or self.engine._has_events:
             self.dispatch.engine_connect(self, self.__branch)
 
+    def schema_for_object(self, obj):
+        """return the schema name for the given schema item taking into
+        account current schema translate map.
+
+        """
+
+        name = obj.schema
+        schema_translate_map = self._schema_translate_map
+
+        if (
+            schema_translate_map
+            and name in schema_translate_map
+            and obj._use_schema_map
+        ):
+            return schema_translate_map[name]
+        else:
+            return name
+
     def _branch(self):
         """Return a new Connection which references this Connection's
         engine and connection; but does not have close_with_result enabled,
@@ -1066,10 +1069,7 @@ class Connection(Connectable):
         dialect = self.dialect
 
         compiled = ddl.compile(
-            dialect=dialect,
-            schema_translate_map=self.schema_for_object
-            if not self.schema_for_object.is_default
-            else None,
+            dialect=dialect, schema_translate_map=self._schema_translate_map
         )
         ret = self._execute_context(
             dialect,
@@ -1103,7 +1103,7 @@ class Connection(Connectable):
                 dialect,
                 elem,
                 tuple(sorted(keys)),
-                self.schema_for_object.hash_key,
+                bool(self._schema_translate_map),
                 len(distilled_params) > 1,
             )
             compiled_sql = self._execution_options["compiled_cache"].get(key)
@@ -1112,9 +1112,7 @@ class Connection(Connectable):
                     dialect=dialect,
                     column_keys=keys,
                     inline=len(distilled_params) > 1,
-                    schema_translate_map=self.schema_for_object
-                    if not self.schema_for_object.is_default
-                    else None,
+                    schema_translate_map=self._schema_translate_map,
                     linting=self.dialect.compiler_linting
                     | compiler.WARN_LINTING,
                 )
@@ -1124,9 +1122,7 @@ class Connection(Connectable):
                 dialect=dialect,
                 column_keys=keys,
                 inline=len(distilled_params) > 1,
-                schema_translate_map=self.schema_for_object
-                if not self.schema_for_object.is_default
-                else None,
+                schema_translate_map=self._schema_translate_map,
                 linting=self.dialect.compiler_linting | compiler.WARN_LINTING,
             )
 
@@ -1974,21 +1970,7 @@ class Engine(Connectable, log.Identified):
     _has_events = False
     _connection_cls = Connection
 
-    schema_for_object = schema._schema_getter(None)
-    """Return the ".schema" attribute for an object.
-
-    Used for :class:`.Table`, :class:`.Sequence` and similar objects,
-    and takes into account
-    the :paramref:`.Connection.execution_options.schema_translate_map`
-    parameter.
-
-      .. versionadded:: 1.1
-
-      .. seealso::
-
-          :ref:`schema_translating`
-
-    """
+    _schema_translate_map = None
 
     def __init__(
         self,
diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index b151b6e48..d0940decf 100644
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -28,7 +28,6 @@ from .. import types as sqltypes
 from .. import util
 from ..sql import compiler
 from ..sql import expression
-from ..sql import schema
 from ..sql.elements import quoted_name
 
 AUTOCOMMIT_REGEXP = re.compile(
@@ -129,6 +128,8 @@ class DefaultDialect(interfaces.Dialect):
 
     server_version_info = None
 
+    default_schema_name = None
+
     construct_arguments = None
     """Optional set of argument specifiers for various SQLAlchemy
     constructs, typically schema items.
@@ -495,20 +496,18 @@ class DefaultDialect(interfaces.Dialect):
                     self._set_connection_isolation(connection, isolation_level)
 
         if "schema_translate_map" in opts:
-            getter = schema._schema_getter(opts["schema_translate_map"])
-            engine.schema_for_object = getter
+            engine._schema_translate_map = map_ = opts["schema_translate_map"]
 
             @event.listens_for(engine, "engine_connect")
             def set_schema_translate_map(connection, branch):
-                connection.schema_for_object = getter
+                connection._schema_translate_map = map_
 
     def set_connection_execution_options(self, connection, opts):
         if "isolation_level" in opts:
             self._set_connection_isolation(connection, opts["isolation_level"])
 
         if "schema_translate_map" in opts:
-            getter = schema._schema_getter(opts["schema_translate_map"])
-            connection.schema_for_object = getter
+            connection._schema_translate_map = opts["schema_translate_map"]
 
     def _set_connection_isolation(self, connection, level):
         if connection.in_transaction():
@@ -701,11 +700,17 @@ class DefaultExecutionContext(interfaces.ExecutionContext):
             self.execution_options = dict(self.execution_options)
             self.execution_options.update(connection._execution_options)
 
+        self.unicode_statement = util.text_type(compiled)
+        if compiled.schema_translate_map:
+            rst = compiled.preparer._render_schema_translates
+            self.unicode_statement = rst(
+                self.unicode_statement, connection._schema_translate_map
+            )
+
         if not dialect.supports_unicode_statements:
-            self.unicode_statement = util.text_type(compiled)
             self.statement = dialect._encoder(self.unicode_statement)[0]
         else:
-            self.statement = self.unicode_statement = util.text_type(compiled)
+            self.statement = self.unicode_statement
 
         self.cursor = self.create_cursor()
         self.compiled_parameters = []
@@ -807,6 +812,12 @@ class DefaultExecutionContext(interfaces.ExecutionContext):
         elif compiled.positional:
             positiontup = self.compiled.positiontup
 
+        if compiled.schema_translate_map:
+            rst = compiled.preparer._render_schema_translates
+            self.unicode_statement = rst(
+                self.unicode_statement, connection._schema_translate_map
+            )
+
         # final self.unicode_statement is now assigned, encode if needed
         # by dialect
         if not dialect.supports_unicode_statements:
diff --git a/lib/sqlalchemy/engine/mock.py b/lib/sqlalchemy/engine/mock.py
index 570ee2d04..bda9e91b5 100644
--- a/lib/sqlalchemy/engine/mock.py
+++ b/lib/sqlalchemy/engine/mock.py
@@ -11,7 +11,6 @@ from . import base
 from . import url as _url
 from .. import util
 from ..sql import ddl
-from ..sql import schema
 
 
 class MockConnection(base.Connectable):
@@ -23,7 +22,8 @@ class MockConnection(base.Connectable):
     dialect = property(attrgetter("_dialect"))
     name = property(lambda s: s._dialect.name)
 
-    schema_for_object = schema._schema_getter(None)
+    def schema_for_object(self, obj):
+        return obj.schema
 
     def connect(self, **kwargs):
         return self
diff --git a/lib/sqlalchemy/engine/reflection.py b/lib/sqlalchemy/engine/reflection.py
index 203369ed8..8ef0d572f 100644
--- a/lib/sqlalchemy/engine/reflection.py
+++ b/lib/sqlalchemy/engine/reflection.py
@@ -701,7 +701,8 @@ class Inspector(object):
 
         dialect = self.bind.dialect
 
-        schema = self.bind.schema_for_object(table)
+        with self._operation_context() as conn:
+            schema = conn.schema_for_object(table)
 
         table_name = table.name