Rewrite query for postgresql Fix rucio#262

vingar · Dec 2, 2017 · ce29f3c · ce29f3c
1 parent 4c6b0b0
commit ce29f3c
Showing 1 changed file with 35 additions and 14 deletions.
diff --git a/lib/rucio/core/replica.py b/lib/rucio/core/replica.py
@@ -1569,19 +1569,41 @@ def get_and_lock_file_replicas_for_dataset(scope, name, nowait=False, restrict_r
     :param session:        The db session in use.
     :returns:              (files in dataset, replicas in dataset)
     """
+    # Get content
+    content_query = session.query(models.DataIdentifierAssociation.child_scope,
+                                  models.DataIdentifierAssociation.child_name,
+                                  models.DataIdentifierAssociation.bytes,
+                                  models.DataIdentifierAssociation.md5,
+                                  models.DataIdentifierAssociation.adler32).\
+        with_hint(models.DataIdentifierAssociation,
+                  "INDEX_RS_ASC(CONTENTS CONTENTS_PK) NO_INDEX_FFS(CONTENTS CONTENTS_PK)",
+                  'oracle').\
+        filter(models.DataIdentifierAssociation.scope == scope,
+               models.DataIdentifierAssociation.name == name)
 
+    files = {}
+    for child_scope, child_name, bytes, md5, adler32 in content_query.yield_per(1000):
+        files[(child_scope, child_name)] = {'scope': child_scope,
+                                            'name': child_name,
+                                            'bytes': bytes,
+                                            'md5': md5,
+                                            'adler32': adler32}
+
+    # Get replicas and lock them
     query = session.query(models.DataIdentifierAssociation.child_scope,
                           models.DataIdentifierAssociation.child_name,
                           models.DataIdentifierAssociation.bytes,
                           models.DataIdentifierAssociation.md5,
                           models.DataIdentifierAssociation.adler32,
                           models.RSEFileAssociation)\
-        .with_hint(models.DataIdentifierAssociation, "INDEX_RS_ASC(CONTENTS CONTENTS_PK) NO_INDEX_FFS(CONTENTS CONTENTS_PK)", 'oracle')\
-        .outerjoin(models.RSEFileAssociation,
-                   and_(models.DataIdentifierAssociation.child_scope == models.RSEFileAssociation.scope,
-                        models.DataIdentifierAssociation.child_name == models.RSEFileAssociation.name,
-                        models.RSEFileAssociation.state != ReplicaState.BEING_DELETED)).\
-        filter(models.DataIdentifierAssociation.scope == scope, models.DataIdentifierAssociation.name == name)
+        .with_hint(models.DataIdentifierAssociation,
+                   "INDEX_RS_ASC(CONTENTS CONTENTS_PK) NO_INDEX_FFS(CONTENTS CONTENTS_PK)",
+                   'oracle')\
+        .filter(and_(models.DataIdentifierAssociation.child_scope == models.RSEFileAssociation.scope,
+                     models.DataIdentifierAssociation.child_name == models.RSEFileAssociation.name,
+                     models.RSEFileAssociation.state != ReplicaState.BEING_DELETED)).\
+        filter(models.DataIdentifierAssociation.scope == scope,
+               models.DataIdentifierAssociation.name == name)
 
     if restrict_rses is not None:
         if len(restrict_rses) < 10:
@@ -1595,20 +1617,19 @@ def get_and_lock_file_replicas_for_dataset(scope, name, nowait=False, restrict_r
                                       models.DataIdentifierAssociation.md5,
                                       models.DataIdentifierAssociation.adler32,
                                       models.RSEFileAssociation)\
-                               .with_hint(models.DataIdentifierAssociation, "INDEX_RS_ASC(CONTENTS CONTENTS_PK) NO_INDEX_FFS(CONTENTS CONTENTS_PK)", 'oracle')\
-                               .outerjoin(models.RSEFileAssociation,
-                                          and_(models.DataIdentifierAssociation.child_scope == models.RSEFileAssociation.scope,
-                                               models.DataIdentifierAssociation.child_name == models.RSEFileAssociation.name,
-                                               models.RSEFileAssociation.state != ReplicaState.BEING_DELETED,
-                                               or_(*rse_clause)))\
+                               .with_hint(models.DataIdentifierAssociation,
+                                          "INDEX_RS_ASC(CONTENTS CONTENTS_PK) NO_INDEX_FFS(CONTENTS CONTENTS_PK)",
+                                          'oracle')\
+                               .filter(and_(models.DataIdentifierAssociation.child_scope == models.RSEFileAssociation.scope,
+                                            models.DataIdentifierAssociation.child_name == models.RSEFileAssociation.name,
+                                            models.RSEFileAssociation.state != ReplicaState.BEING_DELETED,
+                                            or_(*rse_clause)))\
                                .filter(models.DataIdentifierAssociation.scope == scope,
                                        models.DataIdentifierAssociation.name == name)
 
     query = query.with_for_update(nowait=nowait, of=models.RSEFileAssociation.lock_cnt)
 
-    files = {}
     replicas = {}
-
     for child_scope, child_name, bytes, md5, adler32, replica in query:
         if (child_scope, child_name) not in files:
             files[(child_scope, child_name)] = {'scope': child_scope,