@transactional_session
def get_cleaned_updated_collection_replicas(total_workers, worker_number, limit=None, session=None):
    '\n    Get update request for collection replicas.\n    :param total_workers:      Number of total workers.\n    :param worker_number:      id of the executing worker.\n    :param limit:              Maximum numberws to return.\n    :param session:            Database session in use.\n    :returns:                  List of update requests for collection replicas.\n    '
    session.query(models.UpdatedCollectionReplica).filter((models.UpdatedCollectionReplica.rse_id.is_(None) & (~ exists().where(and_((models.CollectionReplica.name == models.UpdatedCollectionReplica.name), (models.CollectionReplica.scope == models.UpdatedCollectionReplica.scope)))))).delete(synchronize_session=False)
    session.query(models.UpdatedCollectionReplica).filter((models.UpdatedCollectionReplica.rse_id.isnot(None) & (~ exists().where(and_((models.CollectionReplica.name == models.UpdatedCollectionReplica.name), (models.CollectionReplica.scope == models.UpdatedCollectionReplica.scope), (models.CollectionReplica.rse_id == models.UpdatedCollectionReplica.rse_id)))))).delete(synchronize_session=False)
    if (session.bind.dialect.name == 'oracle'):
        schema = ''
        if BASE.metadata.schema:
            schema = (BASE.metadata.schema + '.')
        session.execute('DELETE FROM {schema}updated_col_rep A WHERE A.rowid > ANY (SELECT B.rowid FROM {schema}updated_col_rep B WHERE A.scope = B.scope AND A.name=B.name AND A.did_type=B.did_type AND (A.rse_id=B.rse_id OR (A.rse_id IS NULL and B.rse_id IS NULL)))'.format(schema=schema))
    elif (session.bind.dialect.name == 'mysql'):
        subquery1 = session.query(func.max(models.UpdatedCollectionReplica.id).label('max_id')).group_by(models.UpdatedCollectionReplica.scope, models.UpdatedCollectionReplica.name, models.UpdatedCollectionReplica.rse_id).subquery()
        subquery2 = session.query(subquery1.c.max_id).subquery()
        session.query(models.UpdatedCollectionReplica).filter(models.UpdatedCollectionReplica.id.notin_(subquery2)).delete(synchronize_session=False)
    else:
        replica_update_requests = session.query(models.UpdatedCollectionReplica)
        update_requests_with_rse_id = []
        update_requests_without_rse_id = []
        duplicate_request_ids = []
        for update_request in replica_update_requests.all():
            if (update_request.rse_id is not None):
                small_request = {'name': update_request.name, 'scope': update_request.scope, 'rse_id': update_request.rse_id}
                if (small_request not in update_requests_with_rse_id):
                    update_requests_with_rse_id.append(small_request)
                else:
                    duplicate_request_ids.append(update_request.id)
                    continue
            else:
                small_request = {'name': update_request.name, 'scope': update_request.scope}
                if (small_request not in update_requests_without_rse_id):
                    update_requests_without_rse_id.append(small_request)
                else:
                    duplicate_request_ids.append(update_request.id)
                    continue
        for chunk in chunks(duplicate_request_ids, 100):
            session.query(models.UpdatedCollectionReplica).filter(models.UpdatedCollectionReplica.id.in_(chunk)).delete(synchronize_session=False)
    query = session.query(models.UpdatedCollectionReplica)
    if limit:
        query = query.limit(limit)
    return [update_request.to_dict() for update_request in query.all()]
Get update request for collection replicas.
:param total_workers:      Number of total workers.
:param worker_number:      id of the executing worker.
:param limit:              Maximum numberws to return.
:param session:            Database session in use.
:returns:                  List of update requests for collection replicas.
@transactional_session
def get_cleaned_updated_collection_replicas(total_workers, worker_number, limit=None, session=None):
    '\n    Get update request for collection replicas.\n    :param total_workers:      Number of total workers.\n    :param worker_number:      id of the executing worker.\n    :param limit:              Maximum numberws to return.\n    :param session:            Database session in use.\n    :returns:                  List of update requests for collection replicas.\n    '
    session.query(models.UpdatedCollectionReplica).filter((models.UpdatedCollectionReplica.rse_id.is_(None) & (~ exists().where(and_((models.CollectionReplica.name == models.UpdatedCollectionReplica.name), (models.CollectionReplica.scope == models.UpdatedCollectionReplica.scope)))))).delete(synchronize_session=False)
    session.query(models.UpdatedCollectionReplica).filter((models.UpdatedCollectionReplica.rse_id.isnot(None) & (~ exists().where(and_((models.CollectionReplica.name == models.UpdatedCollectionReplica.name), (models.CollectionReplica.scope == models.UpdatedCollectionReplica.scope), (models.CollectionReplica.rse_id == models.UpdatedCollectionReplica.rse_id)))))).delete(synchronize_session=False)
    if (session.bind.dialect.name == 'oracle'):
        schema = 
        if BASE.metadata.schema:
            schema = (BASE.metadata.schema + '.')
        session.execute('DELETE FROM {schema}updated_col_rep A WHERE A.rowid > ANY (SELECT B.rowid FROM {schema}updated_col_rep B WHERE A.scope = B.scope AND A.name=B.name AND A.did_type=B.did_type AND (A.rse_id=B.rse_id OR (A.rse_id IS NULL and B.rse_id IS NULL)))'.format(schema=schema))
    elif (session.bind.dialect.name == 'mysql'):
        subquery1 = session.query(func.max(models.UpdatedCollectionReplica.id).label('max_id')).group_by(models.UpdatedCollectionReplica.scope, models.UpdatedCollectionReplica.name, models.UpdatedCollectionReplica.rse_id).subquery()
        subquery2 = session.query(subquery1.c.max_id).subquery()
        session.query(models.UpdatedCollectionReplica).filter(models.UpdatedCollectionReplica.id.notin_(subquery2)).delete(synchronize_session=False)
    else:
        replica_update_requests = session.query(models.UpdatedCollectionReplica)
        update_requests_with_rse_id = []
        update_requests_without_rse_id = []
        duplicate_request_ids = []
        for update_request in replica_update_requests.all():
            if (update_request.rse_id is not None):
                small_request = {'name': update_request.name, 'scope': update_request.scope, 'rse_id': update_request.rse_id}
                if (small_request not in update_requests_with_rse_id):
                    update_requests_with_rse_id.append(small_request)
                else:
                    duplicate_request_ids.append(update_request.id)
                    continue
            else:
                small_request = {'name': update_request.name, 'scope': update_request.scope}
                if (small_request not in update_requests_without_rse_id):
                    update_requests_without_rse_id.append(small_request)
                else:
                    duplicate_request_ids.append(update_request.id)
                    continue
        for chunk in chunks(duplicate_request_ids, 100):
            session.query(models.UpdatedCollectionReplica).filter(models.UpdatedCollectionReplica.id.in_(chunk)).delete(synchronize_session=False)
    query = session.query(models.UpdatedCollectionReplica)
    if limit:
        query = query.limit(limit)
    return [update_request.to_dict() for update_request in query.all()]