Support querying workflows from both main tables and archive.

CHANGELOG.md

@@ -9,6 +9,7 @@
   - BREAKING CHANGE: Remove `ListWorkflowDefinitionResponse.TransitionDelays.immediate` field, it is not used by nFlow anymore.
   - Enable maintenance (archiving and deleting old workflow instances) by default.
   - Enable workflow instance history clean-up (deleting old actions and state variables) by default.
+  - Add support to query also archived workflow instances.
 
 **Details**
 - `nflow-engine`
@@ -24,6 +25,7 @@
     - Remove `WorkflowSettings.immediateTransitionDelay` field. It is not used by nFlow.
   - Maintenance workflow instance is added to nFlow database by default in startup. Instances that have been in final state longer than 45 days are archived. Archived instances that have been in final state longer than one year are deleted. Maintenance is run every night. Use `nflow.maintenance` configuration options to change the defaults before startup, or update the maintenance workflow instance state variables after the instance has been created.
   - Workflow instance actions and state variables that are older than 45 days are automatically cleaned up occasionally when the instance is processed. Use workflow settings to change the default time period (`setHistoryDeletableAfter`) and condition (`setDeleteHistoryCondition`) of the clean-up.
+  - Add support to query also archived workflow instances when not enough non-archived matches are found.
   - Dependency updates
     - logback-classic update to version 1.2.10
       - http://mailman.qos.ch/pipermail/announce/2021/000164.html
@@ -39,6 +41,7 @@
     - slf4j 1.7.33
 - `nflow-rest-api`
   - BREAKING CHANGE: Remove `ListWorkflowDefinitionResponse.TransitionDelays.immediate` field, it is not used by nFlow.
+  - Add `queryArchive=true` query parameter to query also archived workflow instances when not enough non-archived matches are found.
   - Dependency updates
     - swagger 1.6.4
 - `nflow-jetty`
@@ -52,6 +55,7 @@
   - Dependency updates
     - metrics 4.2.7
 - `nflow-explorer`
+  - Query and show also archived workflow instances when not enough non-archived matches are found. Querying and showing archived instances can be disabled in `config.js`.
   - Dependency updates
     - nodejs 16.13.2
     - npm 8.1.2

nflow-engine/src/main/java/io/nflow/engine/internal/dao/MaintenanceDao.java

@@ -1,8 +1,9 @@
 package io.nflow.engine.internal.dao;
 
 import static io.nflow.engine.internal.dao.DaoUtil.ColumnNamesExtractor.columnNamesExtractor;
-import static io.nflow.engine.internal.dao.TablePrefix.ARCHIVE;
-import static io.nflow.engine.internal.dao.TablePrefix.MAIN;
+import static io.nflow.engine.internal.dao.NflowTable.ACTION;
+import static io.nflow.engine.internal.dao.NflowTable.WORKFLOW;
+import static io.nflow.engine.internal.dao.NflowTable.STATE;
 import static java.lang.System.currentTimeMillis;
 import static java.util.stream.Collectors.joining;
 import static java.util.stream.Stream.generate;
@@ -69,8 +70,8 @@ private String getStateColumns() {
     return stateColumns;
   }
 
-  public List<Long> getOldWorkflowIds(TablePrefix table, DateTime before, int maxWorkflows, Set<String> workflowTypes) {
-    StringBuilder sql = new StringBuilder("select id from ").append(table.nameOf("workflow"))
+  public List<Long> getOldWorkflowIds(TableType type, DateTime before, int maxWorkflows, Set<String> workflowTypes) {
+    StringBuilder sql = new StringBuilder("select id from ").append(WORKFLOW.tableFor(type))
         .append(" where next_activation is null and ").append(sqlVariants.dateLtEqDiff("modified", "?"));
     List<Object> args = new ArrayList<>();
     args.add(sqlVariants.toTimestampObject(before));
@@ -85,32 +86,32 @@ public List<Long> getOldWorkflowIds(TablePrefix table, DateTime before, int maxW
   @Transactional
   public int archiveWorkflows(Collection<Long> workflowIds) {
     String workflowIdParams = params(workflowIds);
-    int archivedInstances = archiveTable("workflow", "id", getWorkflowColumns(), workflowIdParams);
-    int archivedActions = archiveTable("workflow_action", "workflow_id", getActionColumns(), workflowIdParams);
-    int archivedStates = archiveTable("workflow_state", "workflow_id", getStateColumns(), workflowIdParams);
+    int archivedInstances = archiveTable(WORKFLOW, "id", getWorkflowColumns(), workflowIdParams);
+    int archivedActions = archiveTable(ACTION, "workflow_id", getActionColumns(), workflowIdParams);
+    int archivedStates = archiveTable(STATE, "workflow_id", getStateColumns(), workflowIdParams);
     logger.info("Archived {} workflow instances, {} actions and {} states.", archivedInstances, archivedActions, archivedStates);
-    deleteWorkflows(MAIN, workflowIdParams);
+    deleteWorkflows(TableType.MAIN, workflowIdParams);
     return archivedInstances;
   }
 
   @Transactional
-  public int deleteWorkflows(TablePrefix table, Collection<Long> workflowIds) {
+  public int deleteWorkflows(TableType type, Collection<Long> workflowIds) {
     String workflowIdParams = params(workflowIds);
-    return deleteWorkflows(table, workflowIdParams);
+    return deleteWorkflows(type, workflowIdParams);
   }
 
-  private int archiveTable(String table, String workflowIdColumn, String columns, String workflowIdParams) {
-    return jdbc.update("insert into " + ARCHIVE.nameOf(table) + "(" + columns + ") " + "select " + columns + " from "
-        + MAIN.nameOf(table) + sqlVariants.withUpdateSkipLocked() + " where " + workflowIdColumn + " in " + workflowIdParams
+  private int archiveTable(NflowTable table, String workflowIdColumn, String columns, String workflowIdParams) {
+    return jdbc.update("insert into " + table.archive + "(" + columns + ") " + "select " + columns + " from " + table.main
+        + sqlVariants.withUpdateSkipLocked() + " where " + workflowIdColumn + " in " + workflowIdParams
         + sqlVariants.forUpdateSkipLocked());
   }
 
-  private int deleteWorkflows(TablePrefix table, String workflowIdParams) {
-    int deletedStates = jdbc.update("delete from " + table.nameOf("workflow_state") + " where workflow_id in " + workflowIdParams);
-    int deletedActions = jdbc.update("delete from " + table.nameOf("workflow_action") + " where workflow_id in " + workflowIdParams);
-    int deletedInstances = jdbc.update("delete from " + table.nameOf("workflow") + " where id in " + workflowIdParams);
+  private int deleteWorkflows(TableType type, String workflowIdParams) {
+    int deletedStates = jdbc.update("delete from " + STATE.tableFor(type) + " where workflow_id in " + workflowIdParams);
+    int deletedActions = jdbc.update("delete from " + ACTION.tableFor(type) + " where workflow_id in " + workflowIdParams);
+    int deletedInstances = jdbc.update("delete from " + WORKFLOW.tableFor(type) + " where id in " + workflowIdParams);
     logger.info("Deleted {} workflow instances, {} actions and {} states from {} tables.", deletedInstances, deletedActions,
-        deletedStates, table.name());
+        deletedStates, type.name());
     return deletedInstances;
   }
 

nflow-engine/src/main/java/io/nflow/engine/internal/dao/NflowTable.java

@@ -0,0 +1,25 @@
+package io.nflow.engine.internal.dao;
+
+import io.nflow.engine.workflow.instance.WorkflowInstance;
+
+public enum NflowTable {
+  WORKFLOW("workflow"),
+  STATE("workflow_state"),
+  ACTION("workflow_action");
+
+  public String main;
+  public String archive;
+
+  NflowTable(String table) {
+    this.main = TableType.MAIN.prefix + table;
+    this.archive = TableType.ARCHIVE.prefix + table;
+  }
+
+  public String tableFor(TableType type) {
+    return type == TableType.MAIN ? main : archive;
+  }
+
+  public String tableFor(WorkflowInstance instance) {
+    return instance.isArchived ? archive : main;
+  }
+}

nflow-engine/src/main/java/io/nflow/engine/internal/dao/TableMetadataChecker.java

@@ -35,8 +35,8 @@ public void ensureCopyingPossible(String sourceTable, String destinationTable) {
     Map<String, ColumnMetadata> sourceMetadataMap = getMetadata(sourceTable);
     Map<String, ColumnMetadata> destMetadataMap = getMetadata(destinationTable);
     if (destMetadataMap.size() < sourceMetadataMap.size()) {
-      throw new IllegalArgumentException(format("Source table %s has more columns than destination table %s", sourceTable,
-          destinationTable));
+      throw new IllegalArgumentException(
+          format("Source table %s has more columns than destination table %s", sourceTable, destinationTable));
     }
     Set<String> sourceKeySet = sourceMetadataMap.keySet();
     Set<String> destKeySet = destMetadataMap.keySet();
@@ -52,7 +52,8 @@ public void ensureCopyingPossible(String sourceTable, String destinationTable) {
       if (!sourceMetadata.typeName.equals(destMetadata.typeName)) {
         throw new IllegalArgumentException(format(
             "Source column %s.%s has type %s and destination column %s.%s has mismatching type %s", sourceTable,
-            sourceMetadata.columnName, sourceMetadata.typeName, destinationTable, destMetadata.columnName, destMetadata.typeName));
+            sourceMetadata.columnName, sourceMetadata.typeName, destinationTable, destMetadata.columnName,
+            destMetadata.typeName));
       }
       if (sourceMetadata.size > destMetadata.size) {
         throw new IllegalArgumentException(format("Source column %s.%s has size %s and destination column %s.%s smaller size %s",

nflow-engine/src/main/java/io/nflow/engine/internal/dao/TableType.java

@@ -0,0 +1,15 @@
+package io.nflow.engine.internal.dao;
+
+public enum TableType {
+  MAIN("nflow_"), ARCHIVE("nflow_archive_");
+
+  public String prefix;
+
+  TableType(String prefix) {
+    this.prefix = prefix;
+  }
+
+  static String convertMainToArchive(String sql) {
+    return sql.replaceAll(MAIN.prefix, ARCHIVE.prefix);
+  }
+}