Merge pull request #4813 from gchq/gh-2334_fix_split_depth_for_stepping

Gh 2334 fix split depth for stepping

Author: stroomdev66

Diff for: stroom-pipeline/src/main/java/stroom/pipeline/factory/PipelineFactory.java

@@ -155,8 +155,22 @@ public Pipeline create(final PipelineData pipelineData,
             throw new PipelineFactoryException("The pipeline has no source element");
         }
 
+        // Get the split depth to use with the stepping controller.
+        int controllerSplitDepth = 1;
+        if (controller != null) {
+            controllerSplitDepth =
+                    getSplitDepth(elementInstances, elementTypeMap, linkSets, sourceElement.getElementId());
+            controllerSplitDepth = Math.max(controllerSplitDepth, 1);
+        }
+
         // Link the instances.
-        link(elementInstances, elementTypeMap, linkSets, controller, sourceElement, sourceElement.getElementId());
+        link(elementInstances,
+                elementTypeMap,
+                linkSets,
+                controller,
+                sourceElement,
+                sourceElement.getElementId(),
+                controllerSplitDepth);
 
         // We need to create a root element that will be a target for the input
         // stream.
@@ -236,12 +250,11 @@ public static void setProperty(final ElementRegistry pipelineElementRegistry,
                             // stepping and have code to insert.
                             if (controller != null) {
                                 final PipelineStepRequest request = controller.getRequest();
-                                if (request.getCode() != null && request.getCode().size() > 0) {
+                                if (request.getCode() != null && !request.getCode().isEmpty()) {
                                     final String code = request.getCode().get(id);
                                     if (code != null) {
-                                        if (elementInstance instanceof SupportsCodeInjection) {
-                                            final SupportsCodeInjection supportsCodeInjection =
-                                                    (SupportsCodeInjection) elementInstance;
+                                        if (elementInstance instanceof
+                                                final SupportsCodeInjection supportsCodeInjection) {
                                             supportsCodeInjection.setInjectedCode(code);
                                         }
                                     }
@@ -308,7 +321,8 @@ private void link(final Map<String, Element> elementInstances,
                       final Map<String, Set<String>> linkSets,
                       final SteppingController controller,
                       final Element parentElement,
-                      final String parentElementId) {
+                      final String parentElementId,
+                      final int controllerSplitDepth) {
         // Get the child elements of the supplied 'from' element id that we want
         // to work with.
         final List<Element> childElements = getChildElements(parentElementId, elementInstances, elementTypeMap,
@@ -335,37 +349,71 @@ private void link(final Map<String, Element> elementInstances,
                     addMonitor(elementId, elementType, childElement, fragment, controller);
 
                 } else {
-                    fragment = insertRecorder(elementId, elementType, fragment, true, controller);
-                    fragment = insertRecorder(elementId, elementType, fragment, false, controller);
+                    fragment = insertRecorder(elementId, elementType, fragment, true, controller,
+                            controllerSplitDepth);
+                    fragment = insertRecorder(elementId, elementType, fragment, false, controller,
+                            controllerSplitDepth);
                     addMonitor(elementId, elementType, childElement, fragment, controller);
                     fragment = insertRecordDetector(elementType, fragment, true, controller);
                     fragment = insertRecordDetector(elementType, fragment, false, controller);
                 }
             }
 
             // Continue to link the children of this child.
-            link(elementInstances, elementTypeMap, linkSets, controller, fragment.getOut(), elementId);
+            link(elementInstances,
+                    elementTypeMap,
+                    linkSets,
+                    controller,
+                    fragment.getOut(),
+                    elementId,
+                    controllerSplitDepth);
 
             // Now set the target of the parent element to be the 'wrapped'
             // child to complete the link.
             if (parentElement != null && parentElement != fragment.getIn()) {
-                if (!(parentElement instanceof HasTargets)) {
+                if (parentElement instanceof final HasTargets hasTargets) {
+                    if (childElement instanceof Target) {
+                        if (fragment.getIn() instanceof final Target target) {
+                            hasTargets.addTarget(target);
+                        }
+                    } else {
+                        throw new PipelineFactoryException("Attempt to link to an element that is not a target: "
+                                                           + parentElement.getElementId() + " > " + elementId);
+                    }
+                } else {
                     throw new PipelineFactoryException(
                             "Attempt to link from an element that cannot target another element: "
-                                    + parentElement.getElementId());
-                }
-                if (!(childElement instanceof Target)) {
-                    throw new PipelineFactoryException("Attempt to link to an element that is not a target: "
-                            + parentElement.getElementId() + " > " + elementId);
+                            + parentElement.getElementId());
                 }
+            }
+        }
+    }
 
-                final HasTargets hasTargets = (HasTargets) parentElement;
-                if (fragment.getIn() instanceof Target) {
-                    final Target target = (Target) fragment.getIn();
-                    hasTargets.addTarget(target);
-                }
+    /**
+     * Get the split depth for use in stepping.
+     */
+    private int getSplitDepth(final Map<String, Element> elementInstances,
+                              final Map<Element, PipelineElementType> elementTypeMap,
+                              final Map<String, Set<String>> linkSets,
+                              final String parentElementId) {
+        // Get the child elements of the supplied 'from' element id that we want
+        // to work with.
+        final List<Element> childElements = getChildElements(parentElementId, elementInstances, elementTypeMap,
+                linkSets, null);
+
+        // Loop over the child elements and link them to the parent.
+        for (final Element childElement : childElements) {
+            if (childElement instanceof final SplitFilter splitFilter) {
+                return splitFilter.getSplitDepth();
+            }
+            final int childSplitDepth =
+                    getSplitDepth(elementInstances, elementTypeMap, linkSets, childElement.getElementId());
+            if (childSplitDepth != -1) {
+                return childSplitDepth;
             }
         }
+
+        return -1;
     }
 
     /**
@@ -390,7 +438,7 @@ private List<Element> getChildElements(final String fromElementId,
         List<Element> childElements = Collections.emptyList();
 
         final Set<String> toElementIdSet = linkSets.get(fromElementId);
-        if (toElementIdSet != null && toElementIdSet.size() > 0) {
+        if (toElementIdSet != null && !toElementIdSet.isEmpty()) {
             childElements = new ArrayList<>(toElementIdSet.size());
 
             for (final String toElementId : toElementIdSet) {
@@ -440,7 +488,8 @@ private Fragment insertRecorder(final String elementId,
                                     final PipelineElementType elementType,
                                     final Fragment fragment,
                                     final boolean input,
-                                    final SteppingController controller) {
+                                    final SteppingController controller,
+                                    final int controllerSplitDepth) {
         Fragment result = fragment;
 
         // Get any filter settings that might be applied to XML output.
@@ -458,29 +507,25 @@ private Fragment insertRecorder(final String elementId,
                 outputRecorder.setElementId(elementId);
                 result = new Fragment(outputRecorder);
 
-            } else if (in instanceof AbstractInputElement) {
-                final AbstractInputElement filter = (AbstractInputElement) in;
+            } else if (in instanceof final AbstractInputElement filter) {
                 final ReaderRecorder recorder = new ReaderRecorder(errorReceiverProxy);
                 recorder.setElementId(elementId);
                 recorder.setTarget(filter);
                 result = new Fragment(recorder, fragment.getOut());
 
-            } else if (in instanceof AbstractReaderElement) {
-                final AbstractReaderElement filter = (AbstractReaderElement) in;
+            } else if (in instanceof final AbstractReaderElement filter) {
                 final ReaderRecorder recorder = new ReaderRecorder(errorReceiverProxy);
                 recorder.setElementId(elementId);
                 recorder.setTarget(filter);
                 result = new Fragment(recorder, fragment.getOut());
 
-            } else if (in instanceof AbstractParser) {
-                final AbstractParser parser = (AbstractParser) in;
+            } else if (in instanceof final AbstractParser parser) {
                 final ReaderRecorder recorder = new ReaderRecorder(errorReceiverProxy);
                 recorder.setElementId(elementId);
                 recorder.setTarget(parser);
                 result = new Fragment(recorder, fragment.getOut());
 
-            } else if (in instanceof XMLFilter && elementType.hasRole(PipelineElementType.ROLE_MUTATOR)) {
-                final XMLFilter filter = (XMLFilter) in;
+            } else if (in instanceof final XMLFilter filter && elementType.hasRole(PipelineElementType.ROLE_MUTATOR)) {
 
                 // Create stepping filter.
                 final SAXEventRecorder recorder = elementFactory.getElementInstance(SAXEventRecorder.class);
@@ -502,27 +547,24 @@ private Fragment insertRecorder(final String elementId,
                 outputRecorder.setElementId(elementId);
                 result = new Fragment(outputRecorder);
 
-            } else if (out instanceof AbstractInputElement) {
-                final AbstractInputElement filter = (AbstractInputElement) out;
+            } else if (out instanceof final AbstractInputElement filter) {
                 final ReaderRecorder recorder = new ReaderRecorder(errorReceiverProxy);
                 recorder.setElementId(elementId);
                 filter.setTarget(recorder);
                 result = new Fragment(fragment.getIn(), recorder);
 
-            } else if (out instanceof AbstractReaderElement) {
-                final AbstractReaderElement filter = (AbstractReaderElement) out;
+            } else if (out instanceof final AbstractReaderElement filter) {
                 final ReaderRecorder recorder = new ReaderRecorder(errorReceiverProxy);
                 recorder.setElementId(elementId);
                 filter.setTarget(recorder);
                 result = new Fragment(fragment.getIn(), recorder);
 
-            } else if (out instanceof AbstractParser) {
-                final AbstractParser parser = (AbstractParser) out;
+            } else if (out instanceof final AbstractParser parser) {
 
                 // Insert a split filter after the parser to split all XML into
                 // single records.
                 final SplitFilter splitFilter = elementFactory.getElementInstance(SplitFilter.class);
-                splitFilter.setSplitDepth(1);
+                splitFilter.setSplitDepth(controllerSplitDepth);
                 splitFilter.setSplitCount(controller.getRequest().getStepSize());
                 parser.setTarget(splitFilter);
 
@@ -537,18 +579,17 @@ private Fragment insertRecorder(final String elementId,
 
                 result = new Fragment(fragment.getIn(), recorder);
 
-            } else if (out instanceof XMLFilter && elementType.hasRole(PipelineElementType.ROLE_WRITER)) {
-                final XMLFilter filter = (XMLFilter) out;
+            } else if (out instanceof final XMLFilter filter && elementType.hasRole(PipelineElementType.ROLE_WRITER)) {
 
                 final OutputRecorder outputRecorder = elementFactory.getElementInstance(OutputRecorder.class);
                 outputRecorder.setElementId(elementId);
                 ((HasTargets) filter).setTarget(outputRecorder);
 
                 result = new Fragment(fragment.getIn(), outputRecorder);
 
-            } else if (out instanceof XMLFilter && (elementType.hasRole(PipelineElementType.ROLE_MUTATOR)
-                    || elementType.hasRole(PipelineElementType.ROLE_VALIDATOR))) {
-                final XMLFilter filter = (XMLFilter) out;
+            } else if (out instanceof final XMLFilter filter &&
+                       (elementType.hasRole(PipelineElementType.ROLE_MUTATOR) ||
+                        elementType.hasRole(PipelineElementType.ROLE_VALIDATOR))) {
 
                 // Create stepping filter.
                 final SAXEventRecorder recorder = elementFactory.getElementInstance(SAXEventRecorder.class);
@@ -629,7 +670,7 @@ private Fragment insertRecordDetector(final PipelineElementType elementType,
         if (!input) {
             if (elementType.hasRole(PipelineElementType.ROLE_PARSER)) {
                 if (controller.getRecordDetector() == null
-                        || !(controller.getRecordDetector() instanceof SAXRecordDetector)) {
+                    || !(controller.getRecordDetector() instanceof SAXRecordDetector)) {
                     final SAXRecordDetector recordDetector = elementFactory.getElementInstance(SAXRecordDetector.class);
                     controller.setRecordDetector(recordDetector);
                     ((HasTargets) fragment.getOut()).setTarget(recordDetector);

Diff for: stroom-pipeline/src/main/java/stroom/pipeline/filter/SplitFilter.java

@@ -479,4 +479,9 @@ public void setStoreLocations(final boolean storeLocations) {
             locationHolder.setStoreLocations(storeLocations);
         }
     }
+
+    public int getSplitDepth() {
+        // Add a fudge in here to cope with legacy depth being 0 based.
+        return splitDepth - 1;
+    }
 }

Diff for: stroom-pipeline/src/test/java/stroom/pipeline/xml/converter/json/JSONTestDataCreator.java

@@ -0,0 +1,83 @@
+package stroom.pipeline.xml.converter.json;
+
+import java.io.IOException;
+import java.io.Writer;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+public class JSONTestDataCreator {
+
+    void writeFile(final Path path, final JsonType[][] structure) throws IOException {
+        try (final Writer writer = Files.newBufferedWriter(path)) {
+            final JsonType[] level = structure[0];
+            for (int i = 0; i < level.length; i++) {
+                final JsonType type = level[i];
+                writeType(writer, structure, 0, type, i + 1);
+                writer.write("\n");
+            }
+        }
+    }
+
+    private void writeType(final Writer writer,
+                           final JsonType[][] structure,
+                           final int depth,
+                           final JsonType type,
+                           final int no) throws IOException {
+        switch (type) {
+            case OBJECT -> writeJsonObject(writer, structure, depth);
+            case ARRAY -> writeJsonArray(writer, structure, depth);
+            case VALUE -> writer.write("\"value" + no + "\"");
+        }
+    }
+
+    private void writeJsonArray(final Writer writer,
+                                final JsonType[][] structure,
+                                final int depth) throws IOException {
+        writer.write("[\n");
+        final int childDepth = depth + 1;
+        if (childDepth < structure.length) {
+            final JsonType[] types = structure[childDepth];
+            for (int i = 0; i < types.length; i++) {
+                writeIndent(writer, childDepth);
+
+                writeType(writer, structure, childDepth, types[i], i + 1);
+
+                if (i + 1 < types.length) {
+                    writer.write(",");
+                }
+                writer.write("\n");
+            }
+        }
+        writeIndent(writer, depth);
+        writer.write("]");
+    }
+
+    private void writeJsonObject(final Writer writer,
+                                 final JsonType[][] structure,
+                                 final int depth) throws IOException {
+        writer.write("{\n");
+        final int childDepth = depth + 1;
+        if (childDepth < structure.length) {
+            final JsonType[] types = structure[childDepth];
+            for (int i = 0; i < types.length; i++) {
+                writeIndent(writer, childDepth);
+
+                writer.write("\"key" + (i + 1) + "\" : ");
+                writeType(writer, structure, childDepth, types[i], i + 1);
+
+                if (i + 1 < types.length) {
+                    writer.write(",");
+                }
+                writer.write("\n");
+            }
+        }
+        writeIndent(writer, depth);
+        writer.write("}");
+    }
+
+    private void writeIndent(final Writer writer, final int depth) throws IOException {
+        for (int i = 0; i < depth; i++) {
+            writer.write("   ");
+        }
+    }
+}

Diff for: stroom-pipeline/src/test/java/stroom/pipeline/xml/converter/json/JsonType.java

@@ -0,0 +1,7 @@
+package stroom.pipeline.xml.converter.json;
+
+public enum JsonType {
+    OBJECT,
+    ARRAY,
+    VALUE
+}