Query hints for update queries
Motivation
----------
For UPDATE queries on oracle the querie hints are missing.

Modification
------------
This might be related to the fact that there are two different ways in sqlalchemy to generate hints:
- [with_hint](https://docs.sqlalchemy.org/en/14/orm/query.html#sqlalchemy.orm.Query.with_hint) which is generally used by us, but according to the documentation should only work for SELECT queries.
- [prefix_with](https://docs.sqlalchemy.org/en/14/core/selectable.html#sqlalchemy.sql.expression.HasPrefixes.prefix_with) whcih can be used for all queries.

Change UPDATES to `prefix_with`.


diff --git a/lib/rucio/core/did.py b/lib/rucio/core/did.py
index 40e3c8b37..9ae5449f8 100644
--- a/lib/rucio/core/did.py
+++ b/lib/rucio/core/did.py
@@ -36,6 +36,7 @@
 # - Vivek Nigam <viveknigam.nigam3@gmail.com>, 2020
 # - Rahul Chauhan <omrahulchauhan@gmail.com>, 2021
 # - Radu Carpa <radu.carpa@cern.ch>, 2021
+# - David Población Criado <david.poblacion.criado@cern.ch>, 2021
 
 import logging
 import random
@@ -45,7 +46,7 @@ from hashlib import md5
 from re import match
 
 from six import string_types
-from sqlalchemy import and_, or_, exists
+from sqlalchemy import and_, or_, exists, update, delete
 from sqlalchemy.exc import DatabaseError, IntegrityError
 from sqlalchemy.orm.exc import NoResultFound
 from sqlalchemy.sql import not_, func
@@ -316,11 +317,13 @@ def __add_files_to_archive(scope, name, files, account, ignore_duplicate=False,
         new_files and session.bulk_insert_mappings(models.DataIdentifier, new_files)
         if existing_files_condition:
             for chunk in chunks(existing_files_condition, 20):
-                session.query(models.DataIdentifier).\\
-                    with_hint(models.DataIdentifier, "INDEX(DIDS DIDS_PK)", 'oracle').\\
-                    filter(models.DataIdentifier.did_type == DIDType.FILE).\\
-                    filter(or_(models.DataIdentifier.constituent.is_(None), models.DataIdentifier.constituent == false())).\\
-                    filter(or_(*chunk)).update({'constituent': True})
+                stmt = update(models.DataIdentifier).\\
+                    prefix_with("/*+ INDEX(DIDS DIDS_PK) */", dialect='oracle').\\
+                    where(models.DataIdentifier.did_type == DIDType.FILE).\\
+                    where(or_(models.DataIdentifier.constituent.is_(None), models.DataIdentifier.constituent == false())).\\
+                    where(or_(*chunk)).\\
+                    values(constituent=True)
+                session.execute(stmt)
         contents and session.bulk_insert_mappings(models.ConstituentAssociation, contents)
         session.flush()
     except IntegrityError as error:
@@ -1710,11 +1713,10 @@ def resurrect(dids, session=None):
             kargs['expired_at'] = None
         kargs.pop("_sa_instance_state", None)
 
-        session.query(models.DeletedDataIdentifier).\\
-            with_hint(models.DeletedDataIdentifier,
-                      "INDEX(DELETED_DIDS DELETED_DIDS_PK)", 'oracle').\\
-            filter_by(scope=did['scope'], name=did['name']).\\
-            delete()
+        stmt = delete(models.DeletedDataIdentifier).\\
+            prefix_with("/*+ INDEX(DELETED_DIDS DELETED_DIDS_PK) */", dialect='oracle').\\
+            filter_by(scope=did['scope'], name=did['name'])
+        session.execute(stmt)
 
         models.DataIdentifier(**kargs).\\
             save(session=session, flush=False)
diff --git a/lib/rucio/core/did_meta_plugins/did_column_meta.py b/lib/rucio/core/did_meta_plugins/did_column_meta.py
index a4721f8a7..74c825aec 100644
--- a/lib/rucio/core/did_meta_plugins/did_column_meta.py
+++ b/lib/rucio/core/did_meta_plugins/did_column_meta.py
@@ -30,11 +30,12 @@
 # - Brandon White, <bjwhite@fnal.gov>, 2019
 # - Aristeidis Fkiaras <aristeidis.fkiaras@cern.ch>, 2020
 # - Benedikt Ziemons <benedikt.ziemons@cern.ch>, 2020-2021
+# - David Población Criado <david.poblacion.criado@cern.ch>, 2021
 
 from datetime import datetime, timedelta
 
 from six import string_types
-from sqlalchemy import or_
+from sqlalchemy import or_, update
 from sqlalchemy.exc import CompileError, InvalidRequestError
 from sqlalchemy.orm.exc import NoResultFound
 from sqlalchemy.sql import func
@@ -215,8 +216,12 @@ class DidColumnMeta(DidMetaPlugin):
 
                 for child_scope, child_name in content_query:
                     try:
-                        child_did_query = session.query(models.DataIdentifier).with_hint(models.DataIdentifier, "INDEX(DIDS DIDS_PK)", 'oracle').filter_by(scope=child_scope, name=child_name)
-                        child_did_query.update(remainder, synchronize_session='fetch')
+                        stmt = update(models.DataIdentifier)\\
+                            .prefix_with("/*+ INDEX(DIDS DIDS_PK) */", dialect='oracle')\\
+                            .filter_by(scope=child_scope, name=child_name)\\
+                            .execution_options(synchronize_session='fetch')\\
+                            .values(remainder)
+                        session.execute(stmt)
                     except CompileError as error:
                         raise exception.InvalidMetadata(error)
                     except InvalidRequestError:
diff --git a/lib/rucio/core/message.py b/lib/rucio/core/message.py
index c1ba93b88..d218dd6e1 100644
--- a/lib/rucio/core/message.py
+++ b/lib/rucio/core/message.py
@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2014-2020 CERN
+# Copyright 2014-2021 CERN
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -22,10 +22,11 @@
 # - Brandon White <bjwhite@fnal.gov>, 2019
 # - Cedric Serfon <cedric.serfon@cern.ch>, 2020
 # - Benedikt Ziemons <benedikt.ziemons@cern.ch>, 2020
+# - David Población Criado <david.poblacion.criado@cern.ch>, 2021
 
 import json
 
-from sqlalchemy import or_
+from sqlalchemy import or_, delete, update
 from sqlalchemy.exc import IntegrityError
 
 from dogpile.cache import make_region
@@ -170,10 +171,11 @@ def delete_messages(messages, session=None):
 
     try:
         if message_condition:
-            session.query(Message).\\
-                with_hint(Message, "index(messages MESSAGES_ID_PK)", 'oracle').\\
-                filter(or_(*message_condition)).\\
-                delete(synchronize_session=False)
+            stmt = delete(Message).\\
+                prefix_with("/*+ index(messages MESSAGES_ID_PK) */", dialect='oracle').\\
+                where(or_(*message_condition)).\\
+                execution_options(synchronize_session=False)
+            session.execute(stmt)
 
             session.bulk_insert_mappings(MessageHistory, messages)
     except IntegrityError as e:
@@ -211,10 +213,12 @@ def update_messages_services(messages, services, session=None):
 
     try:
         if message_condition:
-            session.query(Message).\\
-                with_hint(Message, "index(messages MESSAGES_ID_PK)", 'oracle').\\
-                filter(or_(*message_condition)).\\
-                update({'services': services}, synchronize_session=False)
+            stmt = update(Message).\\
+                prefix_with("/*+ index(messages MESSAGES_ID_PK) */", dialect='oracle').\\
+                where(or_(*message_condition)).\\
+                execution_options(synchronize_session=False).\\
+                values(services=services)
+            session.execute(stmt)
 
     except IntegrityError as err:
         raise RucioException(err.args)
diff --git a/lib/rucio/core/replica.py b/lib/rucio/core/replica.py
index d48754dd0..c509b0a91 100644
--- a/lib/rucio/core/replica.py
+++ b/lib/rucio/core/replica.py
@@ -38,6 +38,7 @@
 # - Benedikt Ziemons <benedikt.ziemons@cern.ch>, 2020-2021
 # - Radu Carpa <radu.carpa@cern.ch>, 2021
 # - Gabriele Fronzé <sucre.91@hotmail.it>, 2021
+# - David Población Criado <david.poblacion.criado@cern.ch>, 2021
 
 from __future__ import print_function
 
@@ -2169,24 +2170,28 @@ def touch_replica(replica, session=None):
             with_hint(models.RSEFileAssociation, "index(REPLICAS REPLICAS_PK)", 'oracle').\\
             with_for_update(nowait=True).one()
 
-        session.query(models.RSEFileAssociation).filter_by(rse_id=replica['rse_id'], scope=replica['scope'], name=replica['name']).\\
-            with_hint(models.RSEFileAssociation, "index(REPLICAS REPLICAS_PK)", 'oracle').\\
-            update({'accessed_at': accessed_at,
-                    'tombstone': case([(and_(models.RSEFileAssociation.tombstone != none_value,
-                                             models.RSEFileAssociation.tombstone != OBSOLETE),
-                                        accessed_at)],
-                                      else_=models.RSEFileAssociation.tombstone)},
-                   synchronize_session=False)
+        stmt = update(models.RSEFileAssociation).\\
+            filter_by(rse_id=replica['rse_id'], scope=replica['scope'], name=replica['name']).\\
+            prefix_with("/*+ index(REPLICAS REPLICAS_PK) */", dialect='oracle').\\
+            execution_options(synchronize_session=False).\\
+            values(accessed_at=accessed_at,
+                   tombstone=case([(and_(models.RSEFileAssociation.tombstone != none_value,
+                                         models.RSEFileAssociation.tombstone != OBSOLETE),
+                                    accessed_at)],
+                                  else_=models.RSEFileAssociation.tombstone))
+        session.execute(stmt)
 
         session.query(models.DataIdentifier).\\
             filter_by(scope=replica['scope'], name=replica['name'], did_type=DIDType.FILE).\\
             with_hint(models.DataIdentifier, "INDEX(DIDS DIDS_PK)", 'oracle').\\
             with_for_update(nowait=True).one()
 
-        session.query(models.DataIdentifier).\\
+        stmt = update(models.DataIdentifier).\\
             filter_by(scope=replica['scope'], name=replica['name'], did_type=DIDType.FILE).\\
-            with_hint(models.DataIdentifier, "INDEX(DIDS DIDS_PK)", 'oracle').\\
-            update({'accessed_at': accessed_at}, synchronize_session=False)
+            prefix_with("/*+ INDEX(DIDS DIDS_PK) */", dialect='oracle').\\
+            execution_options(synchronize_session=False).\\
+            values(accessed_at=accessed_at)
+        session.execute(stmt)
 
     except DatabaseError:
         return False
@@ -3272,7 +3277,7 @@ def set_tombstone(rse_id, scope, name, tombstone=OBSOLETE, session=None):
     :param tombstone: the tombstone to set. Default is OBSOLETE
     :param session: database session in use.
     """
-    rowcount = session.query(models.RSEFileAssociation).filter(
+    stmt = update(models.RSEFileAssociation).where(
         and_(
             models.RSEFileAssociation.rse_id == rse_id,
             models.RSEFileAssociation.name == name,
@@ -3285,11 +3290,13 @@ def set_tombstone(rse_id, scope, name, tombstone=OBSOLETE, session=None):
                 )
             )
         )
-    ) \\
-        .with_hint(models.RSEFileAssociation, "index(REPLICAS REPLICAS_PK)", 'oracle') \\
-        .update({models.RSEFileAssociation.tombstone: tombstone}, synchronize_session=False)
+    )\\
+        .prefix_with("/*+ index(REPLICAS REPLICAS_PK) */", dialect='oracle')\\
+        .execution_options(synchronize_session=False)\\
+        .values(tombstone=tombstone)
+    result = session.execute(stmt)
 
-    if rowcount == 0:
+    if result.rowcount == 0:
         try:
             session.query(models.RSEFileAssociation).filter_by(scope=scope, name=name, rse_id=rse_id).one()
             raise exception.ReplicaIsLocked('Replica %s:%s on RSE %s is locked.' % (scope, name, get_rse_name(rse_id=rse_id, session=session)))
diff --git a/lib/rucio/core/request.py b/lib/rucio/core/request.py
index d6231f4ba..d3e79526e 100644
--- a/lib/rucio/core/request.py
+++ b/lib/rucio/core/request.py
@@ -31,6 +31,7 @@
 # - Radu Carpa <radu.carpa@cern.ch>, 2021
 # - Matt Snyder <msnyder@bnl.gov>, 2021
 # - Sahan Dilshan <32576163+sahandilshan@users.noreply.github.com>, 2021
+# - David Población Criado <david.poblacion.criado@cern.ch>, 2021
 
 import datetime
 import json
@@ -42,7 +43,7 @@ from itertools import filterfalse
 from typing import TYPE_CHECKING
 
 from six import string_types
-from sqlalchemy import and_, or_, func
+from sqlalchemy import and_, or_, func, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.sql.expression import asc, false, true
 
@@ -552,11 +553,13 @@ def touch_requests_by_rule(rule_id, session=None):
     record_counter('core.request.touch_requests_by_rule')
 
     try:
-        session.query(models.Request).with_hint(models.Request, "INDEX(REQUESTS REQUESTS_RULEID_IDX)", 'oracle')\\
+        stmt = update(models.Request).prefix_with("/*+ INDEX(REQUESTS REQUESTS_RULEID_IDX) */", dialect='oracle')\\
                                      .filter_by(rule_id=rule_id)\\
-                                     .filter(models.Request.state.in_([RequestState.FAILED, RequestState.DONE, RequestState.LOST, RequestState.NO_SOURCES, RequestState.ONLY_TAPE_SOURCES]))\\
-                                     .filter(models.Request.updated_at < datetime.datetime.utcnow())\\
-                                     .update({'updated_at': datetime.datetime.utcnow() + datetime.timedelta(minutes=20)}, synchronize_session=False)
+                                     .where(models.Request.state.in_([RequestState.FAILED, RequestState.DONE, RequestState.LOST, RequestState.NO_SOURCES, RequestState.ONLY_TAPE_SOURCES]))\\
+                                     .where(models.Request.updated_at < datetime.datetime.utcnow())\\
+                                     .execution_options(synchronize_session=False)\\
+                                     .values(updated_at=datetime.datetime.utcnow() + datetime.timedelta(minutes=20))
+        session.execute(stmt)
     except IntegrityError as error:
         raise RucioException(error.args)
 
diff --git a/lib/rucio/core/temporary_did.py b/lib/rucio/core/temporary_did.py
index c39ad40f8..89bedf842 100644
--- a/lib/rucio/core/temporary_did.py
+++ b/lib/rucio/core/temporary_did.py
@@ -11,13 +11,14 @@
   - Mario Lassnig, <mario.lassnig@cern.ch>, 2017
   - Andrew Lister <andrew.lister@stfc.ac.uk>, 2019
   - Brandon White <bjwhite@fnal.gov>, 2019-2020
+  - David Población Criado <david.poblacion.criado@cern.ch>, 2021
 
   PY3K COMPATIBLE
 """
 
 from datetime import datetime
 
-from sqlalchemy import and_, or_, func
+from sqlalchemy import and_, or_, func, delete
 from sqlalchemy.sql.expression import case
 
 from rucio.core.did import attach_dids
@@ -156,9 +157,11 @@ def delete_temporary_dids(dids, session=None):
                                  models.TemporaryDataIdentifier.name == did['name']))
 
     if where_clause:
-        return session.query(models.TemporaryDataIdentifier).\\
-            with_hint(models.TemporaryDataIdentifier, "INDEX(tmp_dids TMP_DIDS_PK)", 'oracle').\\
-            filter(or_(*where_clause)).delete(synchronize_session=False)
+        stmt = delete(models.TemporaryDataIdentifier).\\
+            prefix_with("/*+ INDEX(tmp_dids TMP_DIDS_PK) */", dialect='oracle').\\
+            where(or_(*where_clause)).execution_options(synchronize_session=False)
+        result = session.execute(stmt)
+        return result.rowcount
     return
 
 
diff --git a/lib/rucio/core/transfer.py b/lib/rucio/core/transfer.py
index 8637fe1d3..5c32ea3b9 100644
--- a/lib/rucio/core/transfer.py
+++ b/lib/rucio/core/transfer.py
@@ -37,6 +37,7 @@
 # - Radu Carpa <radu.carpa@cern.ch>, 2021
 # - Sahan Dilshan <32576163+sahandilshan@users.noreply.github.com>, 2021
 # - Petr Vokac <petr.vokac@fjfi.cvut.cz>, 2021
+# - David Población Criado <david.poblacion.criado@cern.ch>, 2021
 
 from __future__ import division
 
@@ -51,7 +52,7 @@ from typing import TYPE_CHECKING
 
 from dogpile.cache import make_region
 from dogpile.cache.api import NoValue
-from sqlalchemy import and_
+from sqlalchemy import and_, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.sql.expression import false
 
@@ -859,11 +860,13 @@ def touch_transfer(external_host, transfer_id, session=None):
 
     try:
         # don't touch it if it's already touched in 30 seconds
-        session.query(models.Request).with_hint(models.Request, "INDEX(REQUESTS REQUESTS_EXTERNALID_UQ)", 'oracle')\\
+        stmt = update(models.Request).prefix_with("/*+ INDEX(REQUESTS REQUESTS_EXTERNALID_UQ) */", dialect='oracle')\\
                                      .filter_by(external_id=transfer_id)\\
-                                     .filter(models.Request.state == RequestState.SUBMITTED)\\
-                                     .filter(models.Request.updated_at < datetime.datetime.utcnow() - datetime.timedelta(seconds=30))\\
-                                     .update({'updated_at': datetime.datetime.utcnow()}, synchronize_session=False)
+                                     .where(models.Request.state == RequestState.SUBMITTED)\\
+                                     .where(models.Request.updated_at < datetime.datetime.utcnow() - datetime.timedelta(seconds=30))\\
+                                     .execution_options(synchronize_session=False)\\
+                                     .values(updated_at=datetime.datetime.utcnow())
+        session.execute(stmt)
     except IntegrityError as error:
         raise RucioException(error.args)
 
diff --git a/lib/rucio/core/volatile_replica.py b/lib/rucio/core/volatile_replica.py
index 11c574b06..f8aabc62c 100644
--- a/lib/rucio/core/volatile_replica.py
+++ b/lib/rucio/core/volatile_replica.py
@@ -9,13 +9,14 @@
   Authors:
   - Vincent Garonne, <vincent.garonne@cern.ch>, 2016
   - Andrew Lister, <andrew.lister@stfc.ac.uk>, 2019
+  - David Población Criado <david.poblacion.criado@cern.ch>, 2021
 
   PY3K COMPATIBLE
 """
 
 from datetime import datetime
 
-from sqlalchemy import and_, or_, exists
+from sqlalchemy import and_, or_, exists, update
 from sqlalchemy.orm.exc import NoResultFound
 from sqlalchemy.sql.expression import select
 
@@ -55,10 +56,12 @@ def add_volatile_replicas(rse_id, replicas, session=None):
 
     if replica_clause:
         now = datetime.utcnow()
-        session.query(models.RSEFileAssociation).\\
-            with_hint(models.RSEFileAssociation, "index(REPLICAS REPLICAS_PK)", 'oracle').\\
-            filter(or_(*replica_clause)).\\
-            update({'updated_at': now, 'tombstone': now}, synchronize_session=False)
+        stmt = update(models.RSEFileAssociation).\\
+            prefix_with("/*+ index(REPLICAS REPLICAS_PK) */", dialect='oracle').\\
+            where(or_(*replica_clause)).\\
+            execution_options(synchronize_session=False).\\
+            values(updated_at=now, tombstone=now)
+        session.execute(stmt)
 
     if file_clause:
         file_query = session.query(models.DataIdentifier.scope,
['lib/rucio/core/did.py', 'lib/rucio/core/replica.py', 'lib/rucio/core/transfer.py', 'lib/rucio/core/message.py', 'lib/rucio/core/temporary_did.py', 'lib/rucio/core/did_meta_plugins/did_column_meta.py', 'lib/rucio/core/volatile_replica.py', 'lib/rucio/core/request.py']