Source code

001/*
002 * Licensed under the Apache License, Version 2.0 (the "License");
003 * you may not use this file except in compliance with the License.
004 * You may obtain a copy of the License at
005 *
006 *     http://www.apache.org/licenses/LICENSE-2.0
007 *
008 * Unless required by applicable law or agreed to in writing, software
009 * distributed under the License is distributed on an "AS IS" BASIS,
010 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
011 * See the License for the specific language governing permissions and
012 * limitations under the License.
013 */
014package org.gbif.api.model.pipelines;
015
016import java.util.Collection;
017import java.util.Collections;
018import java.util.HashMap;
019import java.util.HashSet;
020import java.util.LinkedHashSet;
021import java.util.LinkedList;
022import java.util.List;
023import java.util.Map;
024import java.util.Queue;
025import java.util.Set;
026import java.util.function.Function;
027import java.util.function.ToIntFunction;
028import java.util.stream.Collectors;
029
030import static org.gbif.api.model.pipelines.StepType.*;
031
032public class PipelinesWorkflow {
033
034  private PipelinesWorkflow() {
035    // NOP
036  }
037
038  private static final Graph<StepType> OCCURRENCE_WF_GRAPH = new Graph<>();
039  private static final Graph<StepType> EVENT_OCCURRENCE_WF_GRAPH = new Graph<>();
040  private static final Graph<StepType> EVENT_WF_GRAPH = new Graph<>();
041  private static final Graph<StepType> VALIDATON_WF_GRAPH = new Graph<>();
042
043  static {
044    // Pipelines occurrence workflow
045    // 0?
046    OCCURRENCE_WF_GRAPH.addNode(DWCDP_STAGE, DWCDP_TO_VERBATIM);
047    OCCURRENCE_WF_GRAPH.addNode(NFS_TO_HDFS, DWCDP_TO_VERBATIM);
048    // 1
049    OCCURRENCE_WF_GRAPH.addNode(DWCDP_TO_VERBATIM, VERBATIM_TO_IDENTIFIER);
050    OCCURRENCE_WF_GRAPH.addNode(DWCA_TO_VERBATIM, VERBATIM_TO_IDENTIFIER);
051    OCCURRENCE_WF_GRAPH.addNode(XML_TO_VERBATIM, VERBATIM_TO_IDENTIFIER);
052    OCCURRENCE_WF_GRAPH.addNode(ABCD_TO_VERBATIM, VERBATIM_TO_IDENTIFIER);
053    // 2
054    OCCURRENCE_WF_GRAPH.addNode(VERBATIM_TO_IDENTIFIER, VERBATIM_TO_INTERPRETED);
055    // 3
056    OCCURRENCE_WF_GRAPH.addNode(VERBATIM_TO_INTERPRETED, INTERPRETED_TO_INDEX);
057    OCCURRENCE_WF_GRAPH.addNode(VERBATIM_TO_INTERPRETED, HDFS_VIEW);
058    OCCURRENCE_WF_GRAPH.addNode(VERBATIM_TO_INTERPRETED, FRAGMENTER);
059
060    // Pipelines event-occurrence workflow
061    // 0?
062    EVENT_OCCURRENCE_WF_GRAPH.addNode(NFS_TO_HDFS, DWCDP_TO_VERBATIM);
063    EVENT_OCCURRENCE_WF_GRAPH.addNode(DWCDP_STAGE, DWCDP_TO_VERBATIM);
064    // 1
065    EVENT_OCCURRENCE_WF_GRAPH.addNode(DWCDP_TO_VERBATIM, VERBATIM_TO_IDENTIFIER);
066    EVENT_OCCURRENCE_WF_GRAPH.addNode(DWCA_TO_VERBATIM, VERBATIM_TO_IDENTIFIER);
067    // 2
068    EVENT_OCCURRENCE_WF_GRAPH.addNode(VERBATIM_TO_IDENTIFIER, VERBATIM_TO_INTERPRETED);
069    // 3
070    EVENT_OCCURRENCE_WF_GRAPH.addNode(VERBATIM_TO_INTERPRETED, INTERPRETED_TO_INDEX);
071    EVENT_OCCURRENCE_WF_GRAPH.addNode(VERBATIM_TO_INTERPRETED, HDFS_VIEW);
072    EVENT_OCCURRENCE_WF_GRAPH.addNode(VERBATIM_TO_INTERPRETED, FRAGMENTER);
073    EVENT_OCCURRENCE_WF_GRAPH.addNode(VERBATIM_TO_INTERPRETED, EVENTS_VERBATIM_TO_INTERPRETED);
074    // 4
075    EVENT_OCCURRENCE_WF_GRAPH.addNode(EVENTS_VERBATIM_TO_INTERPRETED, EVENTS_INTERPRETED_TO_INDEX);
076    EVENT_OCCURRENCE_WF_GRAPH.addNode(EVENTS_VERBATIM_TO_INTERPRETED, EVENTS_HDFS_VIEW);
077
078    // Pipelines event only workflow
079    // 0?
080    EVENT_WF_GRAPH.addNode(NFS_TO_HDFS, DWCDP_TO_VERBATIM);
081    EVENT_WF_GRAPH.addNode(DWCDP_STAGE, DWCDP_TO_VERBATIM);
082    // 1
083    EVENT_WF_GRAPH.addNode(DWCDP_TO_VERBATIM, EVENTS_VERBATIM_TO_INTERPRETED);
084    EVENT_WF_GRAPH.addNode(DWCA_TO_VERBATIM, EVENTS_VERBATIM_TO_INTERPRETED);
085    // 2
086    EVENT_WF_GRAPH.addNode(EVENTS_VERBATIM_TO_INTERPRETED, EVENTS_INTERPRETED_TO_INDEX);
087    EVENT_WF_GRAPH.addNode(EVENTS_VERBATIM_TO_INTERPRETED, EVENTS_HDFS_VIEW);
088
089    // Pipelines validator workflow
090    // 1
091    VALIDATON_WF_GRAPH.addNode(VALIDATOR_UPLOAD_ARCHIVE, VALIDATOR_VALIDATE_ARCHIVE);
092    // 2
093    VALIDATON_WF_GRAPH.addNode(VALIDATOR_VALIDATE_ARCHIVE, VALIDATOR_DWCA_TO_VERBATIM);
094    VALIDATON_WF_GRAPH.addNode(VALIDATOR_VALIDATE_ARCHIVE, VALIDATOR_XML_TO_VERBATIM);
095    VALIDATON_WF_GRAPH.addNode(VALIDATOR_VALIDATE_ARCHIVE, VALIDATOR_ABCD_TO_VERBATIM);
096    VALIDATON_WF_GRAPH.addNode(VALIDATOR_VALIDATE_ARCHIVE, VALIDATOR_TABULAR_TO_VERBATIM);
097    // 3
098    VALIDATON_WF_GRAPH.addNode(VALIDATOR_DWCA_TO_VERBATIM, VALIDATOR_VERBATIM_TO_INTERPRETED);
099    VALIDATON_WF_GRAPH.addNode(VALIDATOR_XML_TO_VERBATIM, VALIDATOR_VERBATIM_TO_INTERPRETED);
100    VALIDATON_WF_GRAPH.addNode(VALIDATOR_ABCD_TO_VERBATIM, VALIDATOR_VERBATIM_TO_INTERPRETED);
101    VALIDATON_WF_GRAPH.addNode(VALIDATOR_TABULAR_TO_VERBATIM, VALIDATOR_VERBATIM_TO_INTERPRETED);
102    // 4
103    VALIDATON_WF_GRAPH.addNode(VALIDATOR_VERBATIM_TO_INTERPRETED, VALIDATOR_INTERPRETED_TO_INDEX);
104    // 5
105    VALIDATON_WF_GRAPH.addNode(VALIDATOR_INTERPRETED_TO_INDEX, VALIDATOR_COLLECT_METRICS);
106  }
107
108  public static Graph<StepType> getOccurrenceWorkflow() {
109    return OCCURRENCE_WF_GRAPH;
110  }
111
112  public static Graph<StepType> getEventOccurrenceWorkflow() {
113    return EVENT_OCCURRENCE_WF_GRAPH;
114  }
115
116  public static Graph<StepType> getEventWorkflow() {
117    return EVENT_WF_GRAPH;
118  }
119
120  public static Graph<StepType> getValidatorWorkflow() {
121    return VALIDATON_WF_GRAPH;
122  }
123
124  public static Graph<StepType> getWorkflow(boolean containsOccurrences, boolean containsEvents) {
125    if(containsOccurrences && containsEvents){
126      return getEventOccurrenceWorkflow();
127    } else if(containsOccurrences){
128      return getOccurrenceWorkflow();
129    } else if (containsEvents){
130      return getEventWorkflow();
131    }
132    return new Graph<>();
133  }
134
135  public static class Graph<T> {
136
137    public class Edge {
138      private final T node;
139
140      public Edge(T node) {
141        this.node = node;
142      }
143
144      public T getNode() {
145        return node;
146      }
147    }
148
149    private final Map<T, LinkedList<Edge>> nodes = new HashMap<>();
150
151    private final Map<T, Integer> levels = new HashMap<>();
152
153    private final ToIntFunction<T> calculateLevelFn = t -> levels.get(t) != null ? levels.get(t) + 1 : 1;
154
155    private final ToIntFunction<T> findLevelFn = t -> nodes.values()
156      .stream()
157      .flatMap(Collection::stream)
158      .filter(x->x.getNode().equals(t))
159      .findAny()
160      .map(x->levels.get(x.getNode()))
161      .orElse(1);
162
163    public int getNodesQuantity() {
164      return nodes.size();
165    }
166
167    public List<Edge> getNodeEdges(T node) {
168      return nodes.get(node);
169    }
170
171    public Set<T> getAllNodes() {
172      return nodes.keySet();
173    }
174
175    public Set<T> getAllNodesFor(Set<T> fromTypesSet) {
176      return fromTypesSet.stream()
177        .map(ft -> bfs(this, ft))
178        .flatMap(Collection::stream)
179        .collect(Collectors.toSet());
180    }
181
182    /**
183     * Returns the depth level of a node in the workflow graph, calculated relative to the insertion
184     * order of edges in the static graph initializer.
185     *
186     * <p><b>Deprecated:</b> This method is fundamentally unreliable for graphs with multiple entry
187     * points, as there is no globally meaningful notion of depth in a directed acyclic graph with
188     * more than one root. The level of a node is only well-defined relative to a specific starting
189     * node, and will produce incorrect or inconsistent results when new entry points are added to
190     * the graph.
191     *
192     * <p>Use {@link Graph#getRootNodesFor(Set)} to determine which steps should be triggered first
193     * from a given set of requested steps. For execution history and step ordering, prefer deriving
194     * state from the persisted execution record in the registry rather than from static level
195     * arithmetic.
196     *
197     * @param t the node to look up
198     * @return the computed depth level of the node, unreliable in multi-root graphs
199     */
200    @Deprecated
201    public int getLevel(T t) {
202      return this.levels.get(t);
203    }
204
205    public Set<T> getRootNodesFor(Set<T> fromTypesSet) {
206      Map<T, Set<T>> map = fromTypesSet.stream()
207        .collect(Collectors.toMap(Function.identity(), ft -> bfs(this, ft)));
208
209      Set<T> result = new LinkedHashSet<>();
210
211      fromTypesSet.forEach(ts -> {
212        result.add(ts);
213        map.forEach((key, value) -> {
214          if (key != ts && value.contains(ts)) {
215            result.remove(ts);
216          }
217        });
218      });
219
220      return result;
221    }
222
223    private void addNode(T fromNode, T toNode) {
224      if (nodes.containsKey(fromNode)) {
225        LinkedList<Edge> edges = nodes.get(fromNode);
226
227        int order;
228        if (edges.isEmpty()) {
229          order = calculateLevelFn.applyAsInt(fromNode);
230        } else {
231          order = levels.get(edges.getLast().getNode());
232        }
233
234        edges.add(new Edge(toNode));
235        levels.put(toNode, order);
236      } else {
237        int order = calculateLevelFn.applyAsInt(fromNode);
238        nodes.put(fromNode, new LinkedList<>(Collections.singletonList(new Edge(toNode))));
239        levels.put(fromNode, order);
240        levels.put(toNode, order + 1);
241      }
242
243      nodes.computeIfAbsent(toNode, n -> new LinkedList<>(Collections.emptyList()));
244      levels.put(toNode, findLevelFn.applyAsInt(toNode));
245    }
246
247    private static <T> Set<T> bfs(Graph<T> graph, T startNode) {
248
249      Set<T> resultSet = new HashSet<>();
250      Set<T> visitedNodes = new HashSet<>(graph.getNodesQuantity());
251      Queue<T> queue = new LinkedList<>();
252      T currentNode;
253
254      visitedNodes.add(startNode);
255      queue.add(startNode);
256
257      while (!queue.isEmpty()) {
258        currentNode = queue.poll();
259        resultSet.add(currentNode);
260
261        graph.getNodeEdges(currentNode).forEach(edge -> {
262          T node = edge.getNode();
263          if (visitedNodes.contains(node)) {
264            return;
265          }
266          visitedNodes.add(node);
267          queue.add(node);
268        });
269      }
270
271      return resultSet;
272    }
273
274  }
275}