diff --git a/cinder/cmd/manage.py b/cinder/cmd/manage.py
index a983dda6e..6280a3421 100644
--- a/cinder/cmd/manage.py
+++ b/cinder/cmd/manage.py
@@ -60,6 +60,7 @@ import os
 import sys
 
 from oslo_config import cfg
+from oslo_db import exception as db_exc
 from oslo_db.sqlalchemy import migration
 from oslo_log import log as logging
 import oslo_messaging as messaging
@@ -229,7 +230,13 @@ class DbCommands(object):
             print(_("Must supply a positive, non-zero value for age"))
             sys.exit(1)
         ctxt = context.get_admin_context()
-        db.purge_deleted_rows(ctxt, age_in_days)
+
+        try:
+            db.purge_deleted_rows(ctxt, age_in_days)
+        except db_exc.DBReferenceError:
+            print(_("Purge command failed, check cinder-manage "
+                    "logs for more details."))
+            sys.exit(1)
 
 
 class VersionCommands(object):
diff --git a/cinder/db/sqlalchemy/api.py b/cinder/db/sqlalchemy/api.py
index 61b489cc9..d3b30b52f 100644
--- a/cinder/db/sqlalchemy/api.py
+++ b/cinder/db/sqlalchemy/api.py
@@ -4265,9 +4265,10 @@ def purge_deleted_rows(context, age_in_days):
                 result = session.execute(
                     t.delete()
                     .where(t.c.deleted_at < deleted_age))
-        except db_exc.DBReferenceError:
-            LOG.exception(_LE('DBError detected when purging from '
-                              'table=%(table)s'), {'table': table})
+        except db_exc.DBReferenceError as ex:
+            LOG.error(_LE('DBError detected when purging from '
+                          '%(tablename)s: %(error)s.'),
+                      {'tablename': table, 'error': six.text_type(ex)})
             raise
 
         rows_purged = result.rowcount
diff --git a/cinder/tests/unit/db/test_purge.py b/cinder/tests/unit/db/test_purge.py
index e2a07e2fd..c3975352c 100644
--- a/cinder/tests/unit/db/test_purge.py
+++ b/cinder/tests/unit/db/test_purge.py
@@ -18,7 +18,9 @@
 import datetime
 import uuid
 
+from oslo_db import exception as db_exc
 from oslo_utils import timeutils
+from sqlalchemy.dialects import sqlite
 
 from cinder import context
 from cinder import db
@@ -100,3 +102,35 @@ class PurgeDeletedTest(test.TestCase):
         self.assertRaises(exception.InvalidParameterValue,
                           db.purge_deleted_rows, self.context,
                           age_in_days=-1)
+
+    def test_purge_deleted_rows_integrity_failure(self):
+        dialect = self.engine.url.get_dialect()
+        if dialect == sqlite.dialect:
+            # We're seeing issues with foreign key support in SQLite 3.6.20
+            # SQLAlchemy doesn't support it at all with < SQLite 3.6.19
+            # It works fine in SQLite 3.7.
+            # So return early to skip this test if running SQLite < 3.7
+            import sqlite3
+            tup = sqlite3.sqlite_version_info
+            if tup[0] < 3 or (tup[0] == 3 and tup[1] < 7):
+                self.skipTest(
+                    'sqlite version too old for reliable SQLA foreign_keys')
+            self.conn.execute("PRAGMA foreign_keys = ON")
+
+        # add new entry in volume and volume_admin_metadata for
+        # integrity check
+        uuid_str = uuid.uuid4().hex
+        ins_stmt = self.volumes.insert().values(id=uuid_str)
+        self.conn.execute(ins_stmt)
+        ins_stmt = self.vm.insert().values(volume_id=uuid_str)
+        self.conn.execute(ins_stmt)
+
+        # set volume record to deleted 20 days ago
+        old = timeutils.utcnow() - datetime.timedelta(days=20)
+        make_old = self.volumes.update().where(
+            self.volumes.c.id.in_([uuid_str])).values(deleted_at=old)
+        self.conn.execute(make_old)
+
+        # Verify that purge_deleted_rows fails due to Foreign Key constraint
+        self.assertRaises(db_exc.DBReferenceError, db.purge_deleted_rows,
+                          self.context, age_in_days=10)