Repository: rucio/rucio
Issue: Undertaker can not delete did with Generic Metadata

Motivation
----------
Doing 
```
$ rucio add-dataset testing:test_under2
$ rucio add-did-meta --did testing:test_under2 --key MYKEY --value MYVALUE
$ rucio erase testing:test_under2
```
Gives me the following error on undertaker 1.20.3
```
(psycopg2.errors.ForeignKeyViolation) update or delete on table "dids" violates foreign key constraint "DID_META_FK" on table "did_meta"
DETAIL:  Key (scope, name)=(testing, test_under2) is still referenced from table "did_meta".

[SQL: DELETE FROM dids WHERE dids.scope = %(scope_1)s AND dids.name = %(name_1)s AND (dids.did_type = %(did_type_1)s OR dids.did_type = %(did_type_2)s)]
[parameters: {'name_1': 'test_under2', 'did_type_1': 'C', 'scope_1': 'testing', 'did_type_2': 'D'}]
(Background on this error at: http://sqlalche.me/e/gkpj)
2019-08-16 15:31:02,264	7	CRITICAL	Traceback (most recent call last):
  File "/usr/lib/python2.7/site-packages/rucio/daemons/undertaker/undertaker.py", line 109, in undertaker
    logging.error('Undertaker(%s): Got database error %s.', worker_number, str(error))
UnboundLocalError: local variable 'error' referenced before assignment

2019-08-16 15:31:03,276	7	INFO	Undertaker(1/1): Live gives {'nr_threads': 1, 'assign_thread': 0}
2019-08-16 15:31:03,282	7	INFO	Undertaker(1): Receive 1 dids to delete
2019-08-16 15:31:03,282	7	INFO	Removing did testing:test_under2 (DATASET)
(psycopg2.errors.ForeignKeyViolation) update or delete on table "dids" violates foreign key constraint "DID_META_FK" on table "did_meta"
DETAIL:  Key (scope, name)=(testing, test_under2) is still referenced from table "did_meta".

[SQL: DELETE FROM dids WHERE dids.scope = %(scope_1)s AND dids.name = %(name_1)s AND (dids.did_type = %(did_type_1)s OR dids.did_type = %(did_type_2)s)]
[parameters: {'name_1': 'test_under2', 'did_type_1': 'C', 'scope_1': 'testing', 'did_type_2': 'D'}]
(Background on this error at: http://sqlalche.me/e/gkpj)
2019-08-16 15:31:03,298	7	CRITICAL	Traceback (most recent call last):
  File "/usr/lib/python2.7/site-packages/rucio/daemons/undertaker/undertaker.py", line 109, in undertaker
    logging.error('Undertaker(%s): Got database error %s.', worker_number, str(error))
UnboundLocalError: local variable 'error' referenced before assignment

```
Having a look at the code I believe the UnboundLocalError has been already fixed for the next releases.
The Foreign key constraint seems to be there.

Modification
------------
Delete metadata for did which is about to be deleted?

I wouldn't mind having a look and doing a pr



Changed files:
lib/rucio/core/did.py, lib/rucio/tests/test_undertaker.py

diff --git a/lib/rucio/core/did.py b/lib/rucio/core/did.py
index 6ee232fa0..b578f312e 100644
--- a/lib/rucio/core/did.py
+++ b/lib/rucio/core/did.py
@@ -573,6 +573,7 @@ def delete_dids(dids, account, expire_rules=False, session=None):
     collection_replica_clause, file_clause = [], []
     not_purge_replicas = []
     did_followed_clause = []
+    metadata_to_delete = []
 
     for did in dids:
         logging.info('Removing did %(scope)s:%(name)s (%(did_type)s)' % did)
@@ -625,6 +626,13 @@ def delete_dids(dids, account, expire_rules=False, session=None):
         parent_content_clause.append(and_(models.DataIdentifierAssociation.child_scope == did['scope'], models.DataIdentifierAssociation.child_name == did['name']))
         rule_id_clause.append(and_(models.ReplicationRule.scope == did['scope'], models.ReplicationRule.name == did['name']))
 
+        if session.bind.dialect.name == 'oracle':
+            oracle_version = int(session.connection().connection.version.split('.')[0])
+            if oracle_version >= 12:
+                metadata_to_delete.append(and_(models.DidMeta.scope == did['scope'], models.DidMeta.name == did['name']))
+        else:
+            metadata_to_delete.append(and_(models.DidMeta.scope == did['scope'], models.DidMeta.name == did['name']))
+
         # Send message
         add_message('ERASE', {'account': account.external,
                               'scope': did['scope'].external,
@@ -681,6 +689,19 @@ def delete_dids(dids, account, expire_rules=False, session=None):
             rowcount = session.query(models.CollectionReplica).filter(or_(*collection_replica_clause)).\\
                 delete(synchronize_session=False)
 
+    # Remove generic did metadata
+    if metadata_to_delete:
+        if session.bind.dialect.name == 'oracle':
+            oracle_version = int(session.connection().connection.version.split('.')[0])
+            if oracle_version >= 12:
+                with record_timer_block('undertaker.did_meta'):
+                    rowcount = session.query(models.DidMeta).filter(or_(*metadata_to_delete)).\\
+                        delete(synchronize_session=False)
+        else:
+            with record_timer_block('undertaker.did_meta'):
+                rowcount = session.query(models.DidMeta).filter(or_(*metadata_to_delete)).\\
+                    delete(synchronize_session=False)
+
     # remove data identifier
     if existing_parent_dids:
         # Exit method early to give Judge time to remove locks (Otherwise, due to foreign keys, did removal does not work
diff --git a/lib/rucio/tests/test_undertaker.py b/lib/rucio/tests/test_undertaker.py
index 7e8ec76e0..f1fbc81e6 100644
--- a/lib/rucio/tests/test_undertaker.py
+++ b/lib/rucio/tests/test_undertaker.py
@@ -27,7 +27,7 @@ from nose.tools import assert_not_equal
 from rucio.common.types import InternalAccount, InternalScope
 from rucio.common.utils import generate_uuid
 from rucio.core.account_limit import set_local_account_limit
-from rucio.core.did import add_dids, attach_dids, list_expired_dids, get_did
+from rucio.core.did import add_dids, attach_dids, list_expired_dids, get_did, add_did_meta
 from rucio.core.replica import get_replica
 from rucio.core.rule import add_rules, list_rules
 from rucio.core.rse import get_rse_id, add_rse
@@ -65,6 +65,14 @@ class TestUndertaker:
 
         add_dids(dids=dsns1 + dsns2, account=root)
 
+        # Add generic metadata on did
+        test_metadata = {"test_key": "test_value"}
+        try:
+            add_did_meta(tmp_scope, dsns1[0]['name'], test_metadata)
+        except NotImplementedError:
+            # add_did_meta is not Implemented for Oracle < 12
+            pass
+
         replicas = list()
         for dsn in dsns1 + dsns2:
             files = [{'scope': tmp_scope, 'name': 'file_%s' % generate_uuid(),