Allow dropping a schema with a sequence shared by more than one table.

Fixes: #6071
Change-Id: I5c4483abf075622cccb73cb4c4f8c873174b4e32
(cherry picked from commit 158346a98bcaa28d44f34b194aac75cadd76f1ad)

diff --git a/doc/build/changelog/unreleased_13/6071.rst b/doc/build/changelog/unreleased_13/6071.rst
new file mode 100644 (file)
index 0000000..f9a04c9
--- /dev/null
+++ b/doc/build/changelog/unreleased_13/6071.rst
@@ -0,0 +1,5 @@
+.. change::
+    :tags: bug, schema
+    :tickets: 6071
+
+    Allow dropping a schema with a sequence shared by more than one table.

diff --git a/lib/sqlalchemy/sql/ddl.py b/lib/sqlalchemy/sql/ddl.py
index 5f4af084fb19b5553e28f18740887b3c25f2011b..c70c8c6e58e4b586700600b5064ddc6cecdb6088 100644 (file)
--- a/lib/sqlalchemy/sql/ddl.py
+++ b/lib/sqlalchemy/sql/ddl.py
@@ -937,7 +937,7 @@ class SchemaDropper(DDLBase):
         seq_coll = [
             s
             for s in metadata._sequences.values()
-            if s.column is None and self._can_drop_sequence(s)
+            if self._can_drop_sequence(s)
         ]
 
         event_collection = [t for (t, fks) in collection if t is not None]
@@ -953,14 +953,17 @@ class SchemaDropper(DDLBase):
         for table, fkcs in collection:
             if table is not None:
                 self.traverse_single(
-                    table, drop_ok=True, _is_metadata_operation=True
+                    table,
+                    drop_ok=True,
+                    _is_metadata_operation=True,
+                    _ignore_sequences=seq_coll,
                 )
             else:
                 for fkc in fkcs:
                     self.traverse_single(fkc)
 
         for seq in seq_coll:
-            self.traverse_single(seq, drop_ok=True)
+            self.traverse_single(seq, drop_ok=seq.column is None)
 
         metadata.dispatch.after_drop(
             metadata,
@@ -994,7 +997,13 @@ class SchemaDropper(DDLBase):
     def visit_index(self, index):
         self.connection.execute(DropIndex(index))
 
-    def visit_table(self, table, drop_ok=False, _is_metadata_operation=False):
+    def visit_table(
+        self,
+        table,
+        drop_ok=False,
+        _is_metadata_operation=False,
+        _ignore_sequences=[],
+    ):
         if not drop_ok and not self._can_drop_table(table):
             return
 
@@ -1014,7 +1023,10 @@ class SchemaDropper(DDLBase):
         # latest/core/defaults.html#associating-a-sequence-as-the-server-side-
         # default), so have to be dropped after the table is dropped.
         for column in table.columns:
-            if column.default is not None:
+            if (
+                column.default is not None
+                and column.default not in _ignore_sequences
+            ):
                 self.traverse_single(column.default)
 
         table.dispatch.after_drop(

diff --git a/test/requirements.py b/test/requirements.py
index 791f6500eed5d8cf1e0aa21c670ebf895c4bb9bb..0bc86239e5a53471898289cf6dd3b8bc9d161474 100644 (file)
--- a/test/requirements.py
+++ b/test/requirements.py
@@ -386,8 +386,9 @@ class DefaultRequirements(SuiteRequirements):
     def sequences_as_server_defaults(self):
         """Target database must support SEQUENCE as a server side default."""
 
-        return only_on(
-            "postgresql", "doesn't support sequences as a server side default."
+        return self.sequences + only_on(
+            ["postgresql", "mariadb", "oracle >= 18"],
+            "doesn't support sequences as a server side default.",
         )
 
     @property

diff --git a/test/sql/test_sequences.py b/test/sql/test_sequences.py
index 3df92a179048ebd80dc12d6f7af8b951154b30cb..534e79e247073e75e429728b2192033d62928073 100644 (file)
--- a/test/sql/test_sequences.py
+++ b/test/sql/test_sequences.py
@@ -13,6 +13,8 @@ from sqlalchemy.testing import assert_raises_message
 from sqlalchemy.testing import engines
 from sqlalchemy.testing import eq_
 from sqlalchemy.testing import fixtures
+from sqlalchemy.testing import is_false
+from sqlalchemy.testing import is_true
 from sqlalchemy.testing.assertsql import AllOf
 from sqlalchemy.testing.assertsql import CompiledSQL
 from sqlalchemy.testing.assertsql import EachOf
@@ -395,6 +397,43 @@ class SequenceTest(fixtures.TestBase, testing.AssertsCompiledSQL):
         result = connection.execute(t.insert())
         eq_(result.inserted_primary_key, [1])
 
+    @testing.requires.sequences_as_server_defaults
+    @testing.provide_metadata
+    def test_shared_sequence(self, connection):
+        # test case for #6071
+        common_seq = Sequence("common_sequence", metadata=self.metadata)
+        Table(
+            "table_1",
+            self.metadata,
+            Column(
+                "id",
+                Integer,
+                common_seq,
+                server_default=common_seq.next_value(),
+                primary_key=True,
+            ),
+        )
+        Table(
+            "table_2",
+            self.metadata,
+            Column(
+                "id",
+                Integer,
+                common_seq,
+                server_default=common_seq.next_value(),
+                primary_key=True,
+            ),
+        )
+
+        self.metadata.create_all(connection)
+        is_true(self._has_sequence(connection, "common_sequence"))
+        is_true(testing.db.dialect.has_table(connection, "table_1"))
+        is_true(testing.db.dialect.has_table(connection, "table_2"))
+        self.metadata.drop_all(connection)
+        is_false(self._has_sequence(connection, "common_sequence"))
+        is_false(testing.db.dialect.has_table(connection, "table_1"))
+        is_false(testing.db.dialect.has_table(connection, "table_2"))
+
 
 class TableBoundSequenceTest(fixtures.TablesTest):
     __requires__ = ("sequences",)
@@ -535,19 +574,29 @@ class SequenceAsServerDefaultTest(
         with self.sql_execution_asserter(testing.db) as asserter:
             self.metadata.drop_all(checkfirst=False)
 
+        asserter.assert_(
+            AllOf(
+                CompiledSQL("DROP TABLE t_seq_test_2", {}),
+                CompiledSQL("DROP TABLE t_seq_test", {}),
+            ),
+            AllOf(
+                # dropped as part of metadata level
+                CompiledSQL("DROP SEQUENCE t_seq", {}),
+                CompiledSQL("DROP SEQUENCE t_seq_2", {}),
+            ),
+        )
+
+    def test_drop_ordering_single_table(self):
+        with self.sql_execution_asserter(testing.db) as asserter:
+            for table in self.metadata.tables.values():
+                table.drop(testing.db, checkfirst=False)
+
         asserter.assert_(
             AllOf(
                 CompiledSQL("DROP TABLE t_seq_test_2", {}),
                 EachOf(
                     CompiledSQL("DROP TABLE t_seq_test", {}),
-                    CompiledSQL(
-                        "DROP SEQUENCE t_seq",  # dropped as part of t_seq_test
-                        {},
-                    ),
+                    CompiledSQL("DROP SEQUENCE t_seq", {}),
                 ),
-            ),
-            CompiledSQL(
-                "DROP SEQUENCE t_seq_2",  # dropped as part of metadata level
-                {},
-            ),
+            )
         )