Add identifier_preparer per-execution context for schema translates

Fixed bug where the "schema_translate_map" feature failed to be taken into
account for the use case of direct execution of
:class:`_schema.DefaultGenerator` objects such as sequences, which included
the case where they were "pre-executed" in order to generate primary key
values when implicit_returning was disabled.

Fixes: #5929
Change-Id: I3fed1d0af28be5ce9c9bb572524dcc8411633f60

2 files changed, 78 insertions, 1 deletions

diff --git a/lib/sqlalchemy/testing/provision.py b/lib/sqlalchemy/testing/provision.py
index 2fade1c32..a976abee0 100644
--- a/lib/sqlalchemy/testing/provision.py
+++ b/lib/sqlalchemy/testing/provision.py
@@ -262,7 +262,6 @@ def drop_all_schema_objects(cfg, eng):
                     )
 
     util.drop_all_tables(eng, inspector)
-
     if config.requirements.schemas.enabled_for_config(cfg):
         util.drop_all_tables(eng, inspector, schema=cfg.test_schema)
         util.drop_all_tables(eng, inspector, schema=cfg.test_schema_2)
@@ -273,6 +272,16 @@ def drop_all_schema_objects(cfg, eng):
         with eng.begin() as conn:
             for seq in inspector.get_sequence_names():
                 conn.execute(ddl.DropSequence(schema.Sequence(seq)))
+            if config.requirements.schemas.enabled_for_config(cfg):
+                for schema_name in [cfg.test_schema, cfg.test_schema_2]:
+                    for seq in inspector.get_sequence_names(
+                        schema=schema_name
+                    ):
+                        conn.execute(
+                            ddl.DropSequence(
+                                schema.Sequence(seq, schema=schema_name)
+                            )
+                        )
 
 
 @register.init
diff --git a/lib/sqlalchemy/testing/suite/test_sequence.py b/lib/sqlalchemy/testing/suite/test_sequence.py
index 7445ade00..d6747d253 100644
--- a/lib/sqlalchemy/testing/suite/test_sequence.py
+++ b/lib/sqlalchemy/testing/suite/test_sequence.py
@@ -45,6 +45,34 @@ class SequenceTest(fixtures.TablesTest):
             Column("data", String(50)),
         )
 
+        Table(
+            "seq_no_returning",
+            metadata,
+            Column(
+                "id",
+                Integer,
+                Sequence("noret_id_seq"),
+                primary_key=True,
+            ),
+            Column("data", String(50)),
+            implicit_returning=False,
+        )
+
+        if testing.requires.schemas.enabled:
+            Table(
+                "seq_no_returning_sch",
+                metadata,
+                Column(
+                    "id",
+                    Integer,
+                    Sequence("noret_sch_id_seq", schema=config.test_schema),
+                    primary_key=True,
+                ),
+                Column("data", String(50)),
+                implicit_returning=False,
+                schema=config.test_schema,
+            )
+
     def test_insert_roundtrip(self, connection):
         connection.execute(self.tables.seq_pk.insert(), dict(data="some data"))
         self._assert_round_trip(self.tables.seq_pk, connection)
@@ -72,6 +100,46 @@ class SequenceTest(fixtures.TablesTest):
         row = conn.execute(table.select()).first()
         eq_(row, (testing.db.dialect.default_sequence_base, "some data"))
 
+    def test_insert_roundtrip_no_implicit_returning(self, connection):
+        connection.execute(
+            self.tables.seq_no_returning.insert(), dict(data="some data")
+        )
+        self._assert_round_trip(self.tables.seq_no_returning, connection)
+
+    @testing.combinations((True,), (False,), argnames="implicit_returning")
+    @testing.requires.schemas
+    def test_insert_roundtrip_translate(self, connection, implicit_returning):
+
+        seq_no_returning = Table(
+            "seq_no_returning_sch",
+            MetaData(),
+            Column(
+                "id",
+                Integer,
+                Sequence("noret_sch_id_seq", schema="alt_schema"),
+                primary_key=True,
+            ),
+            Column("data", String(50)),
+            implicit_returning=implicit_returning,
+            schema="alt_schema",
+        )
+
+        connection = connection.execution_options(
+            schema_translate_map={"alt_schema": config.test_schema}
+        )
+        connection.execute(seq_no_returning.insert(), dict(data="some data"))
+        self._assert_round_trip(seq_no_returning, connection)
+
+    @testing.requires.schemas
+    def test_nextval_direct_schema_translate(self, connection):
+        seq = Sequence("noret_sch_id_seq", schema="alt_schema")
+        connection = connection.execution_options(
+            schema_translate_map={"alt_schema": config.test_schema}
+        )
+
+        r = connection.execute(seq)
+        eq_(r, testing.db.dialect.default_sequence_base)
+
 
 class SequenceCompilerTest(testing.AssertsCompiledSQL, fixtures.TestBase):
     __requires__ = ("sequences",)