- Added new parameter :paramref:`.orm.mapper.passive_deletes` to

available mapper options.   This allows a DELETE to proceed
for a joined-table inheritance mapping against the base table only,
while allowing for ON DELETE CASCADE to handle deleting the row
from the subclass tables.
fixes #2349

7 files changed, 367 insertions, 2 deletions

diff --git a/doc/build/changelog/changelog_11.rst b/doc/build/changelog/changelog_11.rst
index 23965818e..a90f801f7 100644
--- a/doc/build/changelog/changelog_11.rst
+++ b/doc/build/changelog/changelog_11.rst
@@ -22,6 +22,21 @@
     :version: 1.1.0b1
 
     .. change::
+        :tags: feature, orm
+        :tickets: 2349
+
+        Added new parameter :paramref:`.orm.mapper.passive_deletes` to
+        available mapper options.   This allows a DELETE to proceed
+        for a joined-table inheritance mapping against the base table only,
+        while allowing for ON DELETE CASCADE to handle deleting the row
+        from the subclass tables.
+
+        .. seealso::
+
+            :ref:`change_2349`
+
+
+    .. change::
         :tags: bug, sybase
         :tickets: 2278
 
diff --git a/doc/build/changelog/migration_11.rst b/doc/build/changelog/migration_11.rst
index d687c5c66..d8bb0d641 100644
--- a/doc/build/changelog/migration_11.rst
+++ b/doc/build/changelog/migration_11.rst
@@ -290,6 +290,66 @@ time on the outside of the subquery.
 
 :ticket:`3582`
 
+.. _change_2349:
+
+passive_deletes feature for joined-inheritance mappings
+-------------------------------------------------------
+
+A joined-table inheritance mapping may now allow a DELETE to proceed
+as a result of :meth:`.Session.delete`, which only emits DELETE for the
+base table, and not the subclass table, allowing configured ON DELETE CASCADE
+to take place for the configured foreign keys.  This is configured using
+the :paramref:`.orm.mapper.passive_deletes` option::
+
+    from sqlalchemy import Column, Integer, String, ForeignKey, create_engine
+    from sqlalchemy.orm import Session
+    from sqlalchemy.ext.declarative import declarative_base
+
+    Base = declarative_base()
+
+
+    class A(Base):
+        __tablename__ = "a"
+        id = Column('id', Integer, primary_key=True)
+        type = Column(String)
+
+        __mapper_args__ = {
+            'polymorphic_on': type,
+            'polymorphic_identity': 'a',
+            'passive_deletes': True
+        }
+
+
+    class B(A):
+        __tablename__ = 'b'
+        b_table_id = Column('b_table_id', Integer, primary_key=True)
+        bid = Column('bid', Integer, ForeignKey('a.id', ondelete="CASCADE"))
+        data = Column('data', String)
+
+        __mapper_args__ = {
+            'polymorphic_identity': 'b'
+        }
+
+With the above mapping, the :paramref:`.orm.mapper.passive_deletes` option
+is configured on the base mapper; it takes effect for all non-base mappers
+that are descendants of the mapper with the option set.  A DELETE for
+an object of type ``B`` no longer needs to retrieve the primary key value
+of ``b_table_id`` if unloaded, nor does it need to emit a DELETE statement
+for the table itself::
+
+    session.delete(some_b)
+    session.commit()
+
+Will emit SQL as::
+
+    DELETE FROM a WHERE a.id = %(id)s
+    {'id': 1}
+    COMMIT
+
+As always, the target database must have foreign key support with
+ON DELETE CASCADE enabled.
+
+:ticket:`2349`
 
 .. _change_3630:
 
diff --git a/doc/build/orm/collections.rst b/doc/build/orm/collections.rst
index 7d474ce65..577cd233e 100644
--- a/doc/build/orm/collections.rst
+++ b/doc/build/orm/collections.rst
@@ -150,6 +150,10 @@ instances of ``MyOtherClass`` which are not loaded, SQLAlchemy assumes that
 "ON DELETE CASCADE" rules will ensure that those rows are deleted by the
 database.
 
+.. seealso::
+
+    :paramref:`.orm.mapper.passive_deletes` - similar feature on :func:`.mapper`
+
 .. currentmodule:: sqlalchemy.orm.collections
 .. _custom_collections:
 
diff --git a/doc/build/orm/relationship_persistence.rst b/doc/build/orm/relationship_persistence.rst
index d4fca2c93..597f674ed 100644
--- a/doc/build/orm/relationship_persistence.rst
+++ b/doc/build/orm/relationship_persistence.rst
@@ -1,4 +1,4 @@
-Special Relationship Persistence Patterns
+fSpecial Relationship Persistence Patterns
 =========================================
 
 .. _post_update:
@@ -209,6 +209,13 @@ referential integrity is used.  When using SQLite, referential integrity
 should be enabled, using the configuration described at
 :ref:`sqlite_foreign_keys`.
 
+.. seealso::
+
+    :ref:`passive_deletes` - supporting ON DELETE CASCADE with relationships
+
+    :paramref:`.orm.mapper.passive_updates` - similar feature on :func:`.mapper`
+
+
 Simulating limited ON UPDATE CASCADE without foreign key support
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
diff --git a/lib/sqlalchemy/orm/mapper.py b/lib/sqlalchemy/orm/mapper.py
index 88dadcc22..53afdcb28 100644
--- a/lib/sqlalchemy/orm/mapper.py
+++ b/lib/sqlalchemy/orm/mapper.py
@@ -112,6 +112,7 @@ class Mapper(InspectionAttr):
                  include_properties=None,
                  exclude_properties=None,
                  passive_updates=True,
+                 passive_deletes=False,
                  confirm_deleted_rows=True,
                  eager_defaults=False,
                  legacy_is_orphan=False,
@@ -319,6 +320,40 @@ class Mapper(InspectionAttr):
            ordering for entities.  By default mappers have no pre-defined
            ordering.
 
+        :param passive_deletes: Indicates DELETE behavior of foreign key
+           columns when a joined-table inheritance entity is being deleted.
+           Defaults to ``False`` for a base mapper; for an inheriting mapper,
+           defaults to ``False`` unless the value is set to ``True``
+           on the superclass mapper.
+
+           When ``True``, it is assumed that ON DELETE CASCADE is configured
+           on the foreign key relationships that link this mapper's table
+           to its superclass table, so that when the unit of work attempts
+           to delete the entity, it need only emit a DELETE statement for the
+           superclass table, and not this table.
+
+           When ``False``, a DELETE statement is emitted for this mapper's
+           table individually.  If the primary key attributes local to this
+           table are unloaded, then a SELECT must be emitted in order to
+           validate these attributes; note that the primary key columns
+           of a joined-table subclass are not part of the "primary key" of
+           the object as a whole.
+
+           Note that a value of ``True`` is **always** forced onto the
+           subclass mappers; that is, it's not possible for a superclass
+           to specify passive_deletes without this taking effect for
+           all subclass mappers.
+
+           .. versionadded:: 1.1
+
+           .. seealso::
+
+               :ref:`passive_deletes` - description of similar feature as
+               used with :func:`.relationship`
+
+               :paramref:`.mapper.passive_updates` - supporting ON UPDATE
+               CASCADE for joined-table inheritance mappers
+
         :param passive_updates: Indicates UPDATE behavior of foreign key
            columns when a primary key column changes on a joined-table
            inheritance mapping.   Defaults to ``True``.
@@ -339,6 +374,9 @@ class Mapper(InspectionAttr):
                :ref:`passive_updates` - description of a similar feature as
                used with :func:`.relationship`
 
+               :paramref:`.mapper.passive_deletes` - supporting ON DELETE
+               CASCADE for joined-table inheritance mappers
+
         :param polymorphic_on: Specifies the column, attribute, or
           SQL expression used to determine the target class for an
           incoming row, when inheriting classes are present.
@@ -559,6 +597,7 @@ class Mapper(InspectionAttr):
         self._dependency_processors = []
         self.validators = util.immutabledict()
         self.passive_updates = passive_updates
+        self.passive_deletes = passive_deletes
         self.legacy_is_orphan = legacy_is_orphan
         self._clause_adapter = None
         self._requires_row_aliasing = False
@@ -971,6 +1010,8 @@ class Mapper(InspectionAttr):
             self.inherits._inheriting_mappers.append(self)
             self.base_mapper = self.inherits.base_mapper
             self.passive_updates = self.inherits.passive_updates
+            self.passive_deletes = self.inherits.passive_deletes or \
+                self.passive_deletes
             self._all_tables = self.inherits._all_tables
 
             if self.polymorphic_identity is not None:
@@ -982,7 +1023,7 @@ class Mapper(InspectionAttr):
                         (self.polymorphic_identity,
                          self.polymorphic_map[self.polymorphic_identity],
                          self, self.polymorphic_identity)
-                )
+                    )
                 self.polymorphic_map[self.polymorphic_identity] = self
 
         else:
diff --git a/lib/sqlalchemy/orm/persistence.py b/lib/sqlalchemy/orm/persistence.py
index e6a2c0634..30b39f600 100644
--- a/lib/sqlalchemy/orm/persistence.py
+++ b/lib/sqlalchemy/orm/persistence.py
@@ -241,6 +241,8 @@ def delete_obj(base_mapper, states, uowtransaction):
         mapper = table_to_mapper[table]
         if table not in mapper._pks_by_table:
             continue
+        elif mapper.inherits and mapper.passive_deletes:
+            continue
 
         delete = _collect_delete_commands(base_mapper, uowtransaction,
                                           table, states_to_delete)
diff --git a/test/orm/inheritance/test_basic.py b/test/orm/inheritance/test_basic.py
index 911d4bc5c..341e142de 100644
--- a/test/orm/inheritance/test_basic.py
+++ b/test/orm/inheritance/test_basic.py
@@ -1149,6 +1149,242 @@ class FlushTest(fixtures.MappedTest):
         assert user_roles.count().scalar() == 1
 
 
+class PassiveDeletesTest(fixtures.MappedTest):
+    __requires__ = ('foreign_keys',)
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table(
+            "a", metadata,
+            Column('id', Integer, primary_key=True),
+            Column('type', String(30))
+        )
+        Table(
+            "b", metadata,
+            Column(
+                'id', Integer, ForeignKey('a.id', ondelete="CASCADE"),
+                primary_key=True),
+            Column('data', String(10))
+        )
+
+        Table(
+            "c", metadata,
+            Column('cid', Integer, primary_key=True),
+            Column('bid', ForeignKey('b.id', ondelete="CASCADE"))
+        )
+
+    @classmethod
+    def setup_classes(cls):
+        class A(cls.Basic):
+            pass
+
+        class B(A):
+            pass
+
+        class C(B):
+            pass
+
+    def _fixture(self, a_p=False, b_p=False, c_p=False):
+        A, B, C = self.classes("A", "B", "C")
+        a, b, c = self.tables("a", "b", "c")
+
+        mapper(
+            A, a, passive_deletes=a_p,
+            polymorphic_on=a.c.type, polymorphic_identity='a')
+        mapper(
+            B, b, inherits=A, passive_deletes=b_p, polymorphic_identity='b')
+        mapper(
+            C, c, inherits=B, passive_deletes=c_p, polymorphic_identity='c')
+
+    def test_none(self):
+        A, B, C = self.classes("A", "B", "C")
+        self._fixture()
+
+        s = Session()
+        a1, b1, c1 = A(id=1), B(id=2), C(cid=1, id=3)
+        s.add_all([a1, b1, c1])
+        s.commit()
+
+        # want to see if the 'C' table loads even though
+        # a and b are loaded
+        c1 = s.query(A).filter_by(id=3).first()
+        s.delete(c1)
+        with self.sql_execution_asserter(testing.db) as asserter:
+            s.flush()
+        asserter.assert_(
+            CompiledSQL(
+                "SELECT c.bid AS c_bid, b.data AS b_data, c.cid AS c_cid "
+                "FROM c, b WHERE :param_1 = b.id AND b.id = c.bid",
+                [{'param_1': 3}]
+            ),
+            CompiledSQL(
+                "DELETE FROM c WHERE c.cid = :cid",
+                [{'cid': 1}]
+            ),
+            CompiledSQL(
+                "DELETE FROM b WHERE b.id = :id",
+                [{'id': 3}]
+            ),
+            CompiledSQL(
+                "DELETE FROM a WHERE a.id = :id",
+                [{'id': 3}]
+            )
+        )
+
+    def test_c_only(self):
+        A, B, C = self.classes("A", "B", "C")
+        self._fixture(c_p=True)
+
+        s = Session()
+        a1, b1, c1 = A(id=1), B(id=2), C(cid=1, id=3)
+        s.add_all([a1, b1, c1])
+        s.commit()
+
+        s.delete(a1)
+
+        with self.sql_execution_asserter(testing.db) as asserter:
+            s.flush()
+        asserter.assert_(
+            CompiledSQL(
+                "SELECT a.id AS a_id, a.type AS a_type "
+                "FROM a WHERE a.id = :param_1",
+                [{'param_1': 1}]
+            ),
+            CompiledSQL(
+                "DELETE FROM a WHERE a.id = :id",
+                [{'id': 1}]
+            )
+        )
+
+        b1.id
+        s.delete(b1)
+        with self.sql_execution_asserter(testing.db) as asserter:
+            s.flush()
+        asserter.assert_(
+            CompiledSQL(
+                "DELETE FROM b WHERE b.id = :id",
+                [{'id': 2}]
+            ),
+            CompiledSQL(
+                "DELETE FROM a WHERE a.id = :id",
+                [{'id': 2}]
+            )
+        )
+
+        # want to see if the 'C' table loads even though
+        # a and b are loaded
+        c1 = s.query(A).filter_by(id=3).first()
+        s.delete(c1)
+        with self.sql_execution_asserter(testing.db) as asserter:
+            s.flush()
+        asserter.assert_(
+            CompiledSQL(
+                "DELETE FROM b WHERE b.id = :id",
+                [{'id': 3}]
+            ),
+            CompiledSQL(
+                "DELETE FROM a WHERE a.id = :id",
+                [{'id': 3}]
+            )
+        )
+
+    def test_b_only(self):
+        A, B, C = self.classes("A", "B", "C")
+        self._fixture(b_p=True)
+
+        s = Session()
+        a1, b1, c1 = A(id=1), B(id=2), C(cid=1, id=3)
+        s.add_all([a1, b1, c1])
+        s.commit()
+
+        s.delete(a1)
+
+        with self.sql_execution_asserter(testing.db) as asserter:
+            s.flush()
+        asserter.assert_(
+            CompiledSQL(
+                "SELECT a.id AS a_id, a.type AS a_type "
+                "FROM a WHERE a.id = :param_1",
+                [{'param_1': 1}]
+            ),
+            CompiledSQL(
+                "DELETE FROM a WHERE a.id = :id",
+                [{'id': 1}]
+            )
+        )
+
+        b1.id
+        s.delete(b1)
+        with self.sql_execution_asserter(testing.db) as asserter:
+            s.flush()
+        asserter.assert_(
+            CompiledSQL(
+                "DELETE FROM a WHERE a.id = :id",
+                [{'id': 2}]
+            )
+        )
+
+        c1.id
+        s.delete(c1)
+        with self.sql_execution_asserter(testing.db) as asserter:
+            s.flush()
+        asserter.assert_(
+            CompiledSQL(
+                "DELETE FROM a WHERE a.id = :id",
+                [{'id': 3}]
+            )
+        )
+
+    def test_a_only(self):
+        A, B, C = self.classes("A", "B", "C")
+        self._fixture(a_p=True)
+
+        s = Session()
+        a1, b1, c1 = A(id=1), B(id=2), C(cid=1, id=3)
+        s.add_all([a1, b1, c1])
+        s.commit()
+
+        s.delete(a1)
+
+        with self.sql_execution_asserter(testing.db) as asserter:
+            s.flush()
+        asserter.assert_(
+            CompiledSQL(
+                "SELECT a.id AS a_id, a.type AS a_type "
+                "FROM a WHERE a.id = :param_1",
+                [{'param_1': 1}]
+            ),
+            CompiledSQL(
+                "DELETE FROM a WHERE a.id = :id",
+                [{'id': 1}]
+            )
+        )
+
+        b1.id
+        s.delete(b1)
+        with self.sql_execution_asserter(testing.db) as asserter:
+            s.flush()
+        asserter.assert_(
+            CompiledSQL(
+                "DELETE FROM a WHERE a.id = :id",
+                [{'id': 2}]
+            )
+        )
+
+        # want to see if the 'C' table loads even though
+        # a and b are loaded
+        c1 = s.query(A).filter_by(id=3).first()
+        s.delete(c1)
+        with self.sql_execution_asserter(testing.db) as asserter:
+            s.flush()
+        asserter.assert_(
+            CompiledSQL(
+                "DELETE FROM a WHERE a.id = :id",
+                [{'id': 3}]
+            )
+        )
+
+
 class OptimizedGetOnDeferredTest(fixtures.MappedTest):
     """test that the 'optimized get' path accommodates deferred columns."""