Merge pull request DSpace#11872 from bram-atmire/9622-doi-organiser-dao-pagination

Add DAO-level pagination to DOIOrganiser bulk operations

Author: tdonohue

dspace-api/src/main/java/org/dspace/identifier/DOIServiceImpl.java

@@ -174,6 +174,12 @@ public List<DOI> getDOIsByStatus(Context context, List<Integer> statuses) throws
         return doiDAO.findByStatus(context, statuses);
     }
 
+    @Override
+    public List<DOI> getDOIsByStatus(Context context, List<Integer> statuses, int limit, int offset)
+        throws SQLException {
+        return doiDAO.findByStatus(context, statuses, limit, offset);
+    }
+
     @Override
     public List<DOI> getSimilarDOIsNotInState(Context context, String doiPattern, List<Integer> statuses,
                                               boolean dsoIsNotNull)

dspace-api/src/main/java/org/dspace/identifier/dao/DOIDAO.java

@@ -33,5 +33,18 @@ public List<DOI> findSimilarNotInState(Context context, String doi, List<Integer
 
     public List<DOI> findByStatus(Context context, List<Integer> statuses) throws SQLException;
 
+    /**
+     * Find all DOIs with any of the given statuses, with pagination support.
+     *
+     * @param context  current DSpace session.
+     * @param statuses desired statuses.
+     * @param limit    maximum number of results to return (-1 for unlimited).
+     * @param offset   number of results to skip (-1 for none).
+     * @return matching DOIs ordered by ID.
+     * @throws SQLException passed through.
+     */
+    public List<DOI> findByStatus(Context context, List<Integer> statuses, int limit, int offset)
+        throws SQLException;
+
     public DOI findDOIByDSpaceObject(Context context, DSpaceObject dso) throws SQLException;
 }

dspace-api/src/main/java/org/dspace/identifier/dao/impl/DOIDAOImpl.java

@@ -71,6 +71,12 @@ public DOI findDOIByDSpaceObject(Context context, DSpaceObject dso, List<Integer
 
     @Override
     public List<DOI> findByStatus(Context context, List<Integer> statuses) throws SQLException {
+        return findByStatus(context, statuses, -1, -1);
+    }
+
+    @Override
+    public List<DOI> findByStatus(Context context, List<Integer> statuses, int limit, int offset)
+        throws SQLException {
         CriteriaBuilder criteriaBuilder = getCriteriaBuilder(context);
         CriteriaQuery criteriaQuery = getCriteriaQuery(criteriaBuilder, DOI.class);
         Root<DOI> doiRoot = criteriaQuery.from(DOI.class);
@@ -80,7 +86,8 @@ public List<DOI> findByStatus(Context context, List<Integer> statuses) throws SQ
             orPredicates.add(criteriaBuilder.equal(doiRoot.get(DOI_.status), status));
         }
         criteriaQuery.where(criteriaBuilder.or(orPredicates.toArray(new Predicate[] {})));
-        return list(context, criteriaQuery, false, DOI.class, -1, -1);
+        criteriaQuery.orderBy(criteriaBuilder.asc(doiRoot.get(DOI_.id)));
+        return list(context, criteriaQuery, false, DOI.class, limit, offset);
     }
 
     @Override

dspace-api/src/main/java/org/dspace/identifier/doi/DOIOrganiser.java

@@ -56,6 +56,25 @@ public class DOIOrganiser {
 
     private static final Logger LOG = LogManager.getLogger(DOIOrganiser.class);
 
+    /**
+     * Number of DOIs to fetch per batch during bulk operations.
+     */
+    private static final int BATCH_SIZE = 100;
+
+    /**
+     * Functional interface for a DOI processing operation.
+     */
+    @FunctionalInterface
+    private interface DOIOperation {
+        /**
+         * Process a single DOI.
+         *
+         * @param doi the DOI to process.
+         * @throws Exception if processing fails.
+         */
+        void process(DOI doi) throws Exception;
+    }
+
     private final DOIIdentifierProvider provider;
     private final Context context;
     private boolean quiet;
@@ -221,107 +240,27 @@ public static void runCLI(Context context, DOIOrganiser organiser, String[] args
         }
 
         if (line.hasOption('s')) {
-            try {
-                List<DOI> dois = doiService
-                    .getDOIsByStatus(context, Arrays.asList(DOIIdentifierProvider.TO_BE_RESERVED));
-                if (dois.isEmpty()) {
-                    System.err.println("There are no objects in the database "
-                                           + "that could be reserved.");
-                }
-
-                for (DOI doi : dois) {
-                    doi = context.reloadEntity(doi);
-                    try {
-                        organiser.reserve(doi);
-                        context.commit();
-                    } catch (RuntimeException e) {
-                        System.err.format("DOI %s for object %s reservation failed, skipping:  %s%n",
-                                doi.getDSpaceObject().getID().toString(),
-                                doi.getDoi(), e.getMessage());
-                        context.rollback();
-                    }
-                }
-            } catch (SQLException ex) {
-                System.err.println("Error in database connection:" + ex.getMessage());
-                ex.printStackTrace(System.err);
-            }
+            List<Integer> statuses = Arrays.asList(DOIIdentifierProvider.TO_BE_RESERVED);
+            processBatched(context, doiService, statuses, organiser::reserve, "reservation");
         }
 
         if (line.hasOption('r')) {
-            try {
-                List<DOI> dois = doiService
-                    .getDOIsByStatus(context, Arrays.asList(DOIIdentifierProvider.TO_BE_REGISTERED));
-                if (dois.isEmpty()) {
-                    System.err.println("There are no objects in the database "
-                                           + "that could be registered.");
-                }
-                for (DOI doi : dois) {
-                    doi = context.reloadEntity(doi);
-                    try {
-                        organiser.register(doi);
-                        context.commit();
-                    } catch (SQLException e) {
-                        System.err.format("DOI %s for object %s registration failed, skipping:  %s%n",
-                                doi.getDSpaceObject().getID().toString(),
-                                doi.getDoi(), e.getMessage());
-                        context.rollback();
-                    }
-                }
-            } catch (SQLException ex) {
-                System.err.format("Error in database connection:  %s%n", ex.getMessage());
-                ex.printStackTrace(System.err);
-            } catch (RuntimeException ex) {
-                System.err.format("Error registering DOI identifier:  %s%n", ex.getMessage());
-            }
+            List<Integer> statuses = Arrays.asList(DOIIdentifierProvider.TO_BE_REGISTERED);
+            processBatched(context, doiService, statuses, organiser::register, "registration");
         }
 
         if (line.hasOption('u')) {
-            try {
-                List<DOI> dois = doiService.getDOIsByStatus(context, Arrays.asList(
-                    DOIIdentifierProvider.UPDATE_BEFORE_REGISTRATION,
-                    DOIIdentifierProvider.UPDATE_RESERVED,
-                    DOIIdentifierProvider.UPDATE_REGISTERED));
-                if (dois.isEmpty()) {
-                    System.err.println("There are no objects in the database "
-                                           + "whose metadata needs an update.");
-                }
-
-                for (DOI doi : dois) {
-                    doi = context.reloadEntity(doi);
-                    organiser.update(doi);
-                    context.commit();
-                }
-            } catch (SQLException ex) {
-                System.err.println("Error in database connection:" + ex.getMessage());
-                ex.printStackTrace(System.err);
-            }
+            List<Integer> statuses = Arrays.asList(
+                DOIIdentifierProvider.UPDATE_BEFORE_REGISTRATION,
+                DOIIdentifierProvider.UPDATE_RESERVED,
+                DOIIdentifierProvider.UPDATE_REGISTERED);
+            processBatched(context, doiService, statuses, organiser::update, "update");
         }
 
         if (line.hasOption('d')) {
-            try {
-                List<DOI> dois = doiService
-                    .getDOIsByStatus(context, Arrays.asList(DOIIdentifierProvider.TO_BE_DELETED));
-                if (dois.isEmpty()) {
-                    System.err.println("There are no objects in the database "
-                                           + "that could be deleted.");
-                }
-
-                for (DOI doi : dois) {
-                    doi = context.reloadEntity(doi);
-                    try {
-                        organiser.delete(doi.getDoi());
-                        context.commit();
-                    } catch (SQLException e) {
-                        System.err.format("DOI %s for object %s deletion failed, skipping:  %s%n",
-                                doi.getDSpaceObject().getID().toString(),
-                                doi.getDoi(), e.getMessage());
-                        context.rollback();
-                    }
-                }
-            } catch (SQLException ex) {
-                System.err.println("Error in database connection:" + ex.getMessage());
-                ex.printStackTrace(System.err);
-            }
+            List<Integer> statuses = Arrays.asList(DOIIdentifierProvider.TO_BE_DELETED);
+            processBatched(context, doiService, statuses,
+                           doi -> organiser.delete(doi.getDoi()), "deletion");
         }
 
         if (line.hasOption("reserve-doi")) {
@@ -385,6 +324,58 @@ public static void runCLI(Context context, DOIOrganiser organiser, String[] args
 
     }
 
+    /**
+     * Process all DOIs matching the given statuses in batches.
+     * Each batch queries from offset 0 because successfully processed DOIs change status and
+     * drop out of subsequent queries. Stops when a batch is empty or an entire batch fails
+     * (to prevent infinite loops).
+     *
+     * @param context     current DSpace session.
+     * @param doiService  the DOI service to query.
+     * @param statuses    the statuses to query for.
+     * @param operation   the operation to perform on each DOI.
+     * @param processName a human-readable name for the operation (for logging).
+     */
+    private static void processBatched(Context context, DOIService doiService,
+                                       List<Integer> statuses, DOIOperation operation,
+                                       String processName) {
+        try {
+            List<DOI> batch;
+            boolean firstBatch = true;
+            do {
+                batch = doiService.getDOIsByStatus(context, statuses, BATCH_SIZE, 0);
+                if (firstBatch && batch.isEmpty()) {
+                    System.err.println("There are no objects in the database "
+                                           + "that could be processed for " + processName + ".");
+                }
+                firstBatch = false;
+
+                int succeeded = 0;
+                for (DOI doi : batch) {
+                    doi = context.reloadEntity(doi);
+                    try {
+                        operation.process(doi);
+                        context.commit();
+                        succeeded++;
+                    } catch (Exception e) {
+                        System.err.format("DOI %s %s failed, skipping: %s%n",
+                                          doi.getDoi(), processName, e.getMessage());
+                        context.rollback();
+                    }
+                }
+                // If no DOI in this batch succeeded, stop to prevent an infinite loop.
+                if (!batch.isEmpty() && succeeded == 0) {
+                    System.err.println("Entire batch failed for " + processName
+                                           + ", stopping to prevent infinite loop.");
+                    break;
+                }
+            } while (!batch.isEmpty());
+        } catch (SQLException ex) {
+            System.err.println("Error in database connection: " + ex.getMessage());
+            ex.printStackTrace(System.err);
+        }
+    }
+
     /**
      * list DOIs queued for reservation or registration
      * @param processName   - process name for display

dspace-api/src/main/java/org/dspace/identifier/service/DOIService.java

@@ -141,6 +141,19 @@ public String formatIdentifier(String identifier)
      */
     public List<DOI> getDOIsByStatus(Context context, List<Integer> statuses) throws SQLException;
 
+    /**
+     * Find DOIs that have one of a given set of statuses, with pagination support.
+     *
+     * @param context  current DSpace session.
+     * @param statuses desired statuses.
+     * @param limit    maximum number of results to return (-1 for unlimited).
+     * @param offset   number of results to skip (-1 for none).
+     * @return matching DOIs ordered by ID.
+     * @throws SQLException passed through.
+     */
+    public List<DOI> getDOIsByStatus(Context context, List<Integer> statuses, int limit, int offset)
+        throws SQLException;
+
     /**
      * Find all DOIs that are similar to the specified pattern and not in the
      * specified states.