@Test public void testGraphMultipleLoopOnce() { Set<Step> steps = Sets.newHashSet(); Map<String, Object> loopStepConfigMap = Maps.newHashMap(); loopStepConfigMap.put(LoopStep.MODE_PROPERTY, LoopStep.MODE_PARALLEL); loopStepConfigMap.put(LoopStep.SOURCE_PROPERTY, LoopStep.SOURCE_RANGE); loopStepConfigMap.put(LoopStep.RANGE_START_PROPERTY, 100); loopStepConfigMap.put(LoopStep.RANGE_END_PROPERTY, 100); Config loopStepConfig = ConfigFactory.parseMap(loopStepConfigMap); RefactorStep loopStep = new LoopStep("loop_step"); loopStep.configure(loopStepConfig); steps.add(loopStep); Map<String, Object> step1ConfigMap = Maps.newHashMap(); step1ConfigMap.put(Step.DEPENDENCIES_CONFIG, Lists.newArrayList("loop_step")); Config step1Config = ConfigFactory.parseMap(step1ConfigMap); Step step1 = new BatchStep("step1"); step1.configure(step1Config); steps.add(step1); Map<String, Object> step2ConfigMap = Maps.newHashMap(); step2ConfigMap.put(Step.DEPENDENCIES_CONFIG, Lists.newArrayList("loop_step")); Config step2Config = ConfigFactory.parseMap(step2ConfigMap); Step step2 = new BatchStep("step2"); step2.configure(step2Config); steps.add(step2); Map<String, Object> step3ConfigMap = Maps.newHashMap(); step3ConfigMap.put(Step.DEPENDENCIES_CONFIG, Lists.newArrayList("loop_step", "step1", "step2")); Config step3Config = ConfigFactory.parseMap(step3ConfigMap); Step step3 = new BatchStep("step3"); step3.configure(step3Config); steps.add(step3); Map<String, Object> step4ConfigMap = Maps.newHashMap(); step4ConfigMap.put(Step.DEPENDENCIES_CONFIG, Lists.newArrayList("loop_step", "step3")); Config step4Config = ConfigFactory.parseMap(step4ConfigMap); Step step4 = new BatchStep("step4"); step4.configure(step4Config); steps.add(step4); Map<String, Object> step5ConfigMap = Maps.newHashMap(); step5ConfigMap.put(Step.DEPENDENCIES_CONFIG, Lists.newArrayList("loop_step", "step3")); Config step5Config = ConfigFactory.parseMap(step5ConfigMap); Step step5 = new BatchStep("step5"); step5.configure(step5Config); steps.add(step5); Set<Step> unrolled = loopStep.refactor(steps); assertEquals(unrolled.size(), 6); assertEquals(StepUtils.getStepForName("loop_step", unrolled).get().getDependencyNames(), Sets.newHashSet()); assertEquals(StepUtils.getStepForName("step1_100", unrolled).get().getDependencyNames(), Sets.newHashSet("loop_step")); assertEquals(StepUtils.getStepForName("step2_100", unrolled).get().getDependencyNames(), Sets.newHashSet("loop_step")); assertEquals(StepUtils.getStepForName("step3_100", unrolled).get().getDependencyNames(), Sets.newHashSet("loop_step", "step1_100", "step2_100")); assertEquals(StepUtils.getStepForName("step4_100", unrolled).get().getDependencyNames(), Sets.newHashSet("loop_step", "step3_100")); assertEquals(StepUtils.getStepForName("step5_100", unrolled).get().getDependencyNames(), Sets.newHashSet("loop_step", "step3_100")); }
@Override public Set<Step> refactor(Set<Step> steps) { List<Row> iterationRows = getIterationRows(steps); String mode = getMode(); Set<Step> loopGraphSteps = StepUtils.getImmediateDependentSteps(this, steps); LOG.debug("Loop graph steps found: " + StepUtils.stepNamesAsString(loopGraphSteps)); Set<Step> loopGraphStepDependents = Sets.newHashSet(); for (Step loopGraphStep : loopGraphSteps) { for (Step loopGraphStepDependentCandidate : StepUtils.getImmediateDependentSteps(loopGraphStep, steps)) { if (!loopGraphSteps.contains(loopGraphStepDependentCandidate)) { loopGraphStepDependents.add(loopGraphStepDependentCandidate); } } } LOG.debug("Loop graph step dependents found: " + StepUtils.stepNamesAsString(loopGraphStepDependents)); steps.removeAll(loopGraphSteps); Set<Step> previousIterationSteps = null; for (Row iterationRow : iterationRows) { String suffix = getSuffix(iterationRow) ; LOG.debug("Constructing loop for iteration value: {}", suffix); Set<Step> iterationSteps = StepUtils.copySteps(loopGraphSteps); for (Step iterationStep : iterationSteps) { LOG.debug("Adjusting dependencies for iteration step: " + iterationStep.getName()); Set<String> dependencies = iterationStep.getDependencyNames(); Set<String> dependenciesToAdd = Sets.newHashSet(); Set<String> dependenciesToRemove = Sets.newHashSet(); for (String dependency : dependencies) { if (StepUtils.getStepForName(dependency, loopGraphSteps).isPresent()) { dependenciesToAdd.add(dependency + "_" + suffix); dependenciesToRemove.add(dependency); } } dependencies.addAll(dependenciesToAdd); LOG.debug("Added dependencies: " + dependenciesToAdd); dependencies.removeAll(dependenciesToRemove); LOG.debug("Removed dependencies: " + dependenciesToRemove); } for (Step iterationStep : iterationSteps) { String adjustedIterationStepName = iterationStep.getName() + "_" + suffix; LOG.debug("Renaming iteration step {} to {}", iterationStep.getName(), adjustedIterationStepName); iterationStep.setName(adjustedIterationStepName); for (Step loopGraphStepDependent : loopGraphStepDependents) { if (!(iterationStep instanceof LoopStep)) { LOG.debug("Adding dependency {} to {}", iterationStep.getName(), loopGraphStepDependent.getName()); loopGraphStepDependent.getDependencyNames().add(adjustedIterationStepName); } } Config parameterSubstConfig = performSubstitutions(iterationStep.getConfig(), iterationRow); iterationStep.setConfig(parameterSubstConfig); LOG.debug("Parameter substitutions completed: {} ", parameterSubstConfig.toString()); } if (mode.equals(MODE_SERIAL) && previousIterationSteps != null) { LOG.debug("Adjusting iteration steps to be dependent on previous iteration's steps"); for (Step iterationStep : iterationSteps) { for (Step previousIterationStep : previousIterationSteps) { if (!(previousIterationStep instanceof LoopStep)) { LOG.debug("Adding dependency {} to iteration step {}", previousIterationStep.getName(), iterationStep.getName()); iterationStep.getDependencyNames().add(previousIterationStep.getName()); } } } } previousIterationSteps = iterationSteps; steps.addAll(iterationSteps); } for (Step loopGraphStepDependent : loopGraphStepDependents) { for (Step loopGraphStep : loopGraphSteps) { loopGraphStepDependent.getDependencyNames().remove(loopGraphStep.getName()); } } LOG.debug("Unrolled steps: " + StepUtils.stepNamesAsString(steps)); this.setState(StepState.FINISHED); return steps; }
LoopStep extends RefactorStep implements ProvidesValidations { @Override public Set<Step> refactor(Set<Step> steps) { List<Row> iterationRows = getIterationRows(steps); String mode = getMode(); Set<Step> loopGraphSteps = StepUtils.getImmediateDependentSteps(this, steps); LOG.debug("Loop graph steps found: " + StepUtils.stepNamesAsString(loopGraphSteps)); Set<Step> loopGraphStepDependents = Sets.newHashSet(); for (Step loopGraphStep : loopGraphSteps) { for (Step loopGraphStepDependentCandidate : StepUtils.getImmediateDependentSteps(loopGraphStep, steps)) { if (!loopGraphSteps.contains(loopGraphStepDependentCandidate)) { loopGraphStepDependents.add(loopGraphStepDependentCandidate); } } } LOG.debug("Loop graph step dependents found: " + StepUtils.stepNamesAsString(loopGraphStepDependents)); steps.removeAll(loopGraphSteps); Set<Step> previousIterationSteps = null; for (Row iterationRow : iterationRows) { String suffix = getSuffix(iterationRow) ; LOG.debug("Constructing loop for iteration value: {}", suffix); Set<Step> iterationSteps = StepUtils.copySteps(loopGraphSteps); for (Step iterationStep : iterationSteps) { LOG.debug("Adjusting dependencies for iteration step: " + iterationStep.getName()); Set<String> dependencies = iterationStep.getDependencyNames(); Set<String> dependenciesToAdd = Sets.newHashSet(); Set<String> dependenciesToRemove = Sets.newHashSet(); for (String dependency : dependencies) { if (StepUtils.getStepForName(dependency, loopGraphSteps).isPresent()) { dependenciesToAdd.add(dependency + "_" + suffix); dependenciesToRemove.add(dependency); } } dependencies.addAll(dependenciesToAdd); LOG.debug("Added dependencies: " + dependenciesToAdd); dependencies.removeAll(dependenciesToRemove); LOG.debug("Removed dependencies: " + dependenciesToRemove); } for (Step iterationStep : iterationSteps) { String adjustedIterationStepName = iterationStep.getName() + "_" + suffix; LOG.debug("Renaming iteration step {} to {}", iterationStep.getName(), adjustedIterationStepName); iterationStep.setName(adjustedIterationStepName); for (Step loopGraphStepDependent : loopGraphStepDependents) { if (!(iterationStep instanceof LoopStep)) { LOG.debug("Adding dependency {} to {}", iterationStep.getName(), loopGraphStepDependent.getName()); loopGraphStepDependent.getDependencyNames().add(adjustedIterationStepName); } } Config parameterSubstConfig = performSubstitutions(iterationStep.getConfig(), iterationRow); iterationStep.setConfig(parameterSubstConfig); LOG.debug("Parameter substitutions completed: {} ", parameterSubstConfig.toString()); } if (mode.equals(MODE_SERIAL) && previousIterationSteps != null) { LOG.debug("Adjusting iteration steps to be dependent on previous iteration's steps"); for (Step iterationStep : iterationSteps) { for (Step previousIterationStep : previousIterationSteps) { if (!(previousIterationStep instanceof LoopStep)) { LOG.debug("Adding dependency {} to iteration step {}", previousIterationStep.getName(), iterationStep.getName()); iterationStep.getDependencyNames().add(previousIterationStep.getName()); } } } } previousIterationSteps = iterationSteps; steps.addAll(iterationSteps); } for (Step loopGraphStepDependent : loopGraphStepDependents) { for (Step loopGraphStep : loopGraphSteps) { loopGraphStepDependent.getDependencyNames().remove(loopGraphStep.getName()); } } LOG.debug("Unrolled steps: " + StepUtils.stepNamesAsString(steps)); this.setState(StepState.FINISHED); return steps; } }
LoopStep extends RefactorStep implements ProvidesValidations { @Override public Set<Step> refactor(Set<Step> steps) { List<Row> iterationRows = getIterationRows(steps); String mode = getMode(); Set<Step> loopGraphSteps = StepUtils.getImmediateDependentSteps(this, steps); LOG.debug("Loop graph steps found: " + StepUtils.stepNamesAsString(loopGraphSteps)); Set<Step> loopGraphStepDependents = Sets.newHashSet(); for (Step loopGraphStep : loopGraphSteps) { for (Step loopGraphStepDependentCandidate : StepUtils.getImmediateDependentSteps(loopGraphStep, steps)) { if (!loopGraphSteps.contains(loopGraphStepDependentCandidate)) { loopGraphStepDependents.add(loopGraphStepDependentCandidate); } } } LOG.debug("Loop graph step dependents found: " + StepUtils.stepNamesAsString(loopGraphStepDependents)); steps.removeAll(loopGraphSteps); Set<Step> previousIterationSteps = null; for (Row iterationRow : iterationRows) { String suffix = getSuffix(iterationRow) ; LOG.debug("Constructing loop for iteration value: {}", suffix); Set<Step> iterationSteps = StepUtils.copySteps(loopGraphSteps); for (Step iterationStep : iterationSteps) { LOG.debug("Adjusting dependencies for iteration step: " + iterationStep.getName()); Set<String> dependencies = iterationStep.getDependencyNames(); Set<String> dependenciesToAdd = Sets.newHashSet(); Set<String> dependenciesToRemove = Sets.newHashSet(); for (String dependency : dependencies) { if (StepUtils.getStepForName(dependency, loopGraphSteps).isPresent()) { dependenciesToAdd.add(dependency + "_" + suffix); dependenciesToRemove.add(dependency); } } dependencies.addAll(dependenciesToAdd); LOG.debug("Added dependencies: " + dependenciesToAdd); dependencies.removeAll(dependenciesToRemove); LOG.debug("Removed dependencies: " + dependenciesToRemove); } for (Step iterationStep : iterationSteps) { String adjustedIterationStepName = iterationStep.getName() + "_" + suffix; LOG.debug("Renaming iteration step {} to {}", iterationStep.getName(), adjustedIterationStepName); iterationStep.setName(adjustedIterationStepName); for (Step loopGraphStepDependent : loopGraphStepDependents) { if (!(iterationStep instanceof LoopStep)) { LOG.debug("Adding dependency {} to {}", iterationStep.getName(), loopGraphStepDependent.getName()); loopGraphStepDependent.getDependencyNames().add(adjustedIterationStepName); } } Config parameterSubstConfig = performSubstitutions(iterationStep.getConfig(), iterationRow); iterationStep.setConfig(parameterSubstConfig); LOG.debug("Parameter substitutions completed: {} ", parameterSubstConfig.toString()); } if (mode.equals(MODE_SERIAL) && previousIterationSteps != null) { LOG.debug("Adjusting iteration steps to be dependent on previous iteration's steps"); for (Step iterationStep : iterationSteps) { for (Step previousIterationStep : previousIterationSteps) { if (!(previousIterationStep instanceof LoopStep)) { LOG.debug("Adding dependency {} to iteration step {}", previousIterationStep.getName(), iterationStep.getName()); iterationStep.getDependencyNames().add(previousIterationStep.getName()); } } } } previousIterationSteps = iterationSteps; steps.addAll(iterationSteps); } for (Step loopGraphStepDependent : loopGraphStepDependents) { for (Step loopGraphStep : loopGraphSteps) { loopGraphStepDependent.getDependencyNames().remove(loopGraphStep.getName()); } } LOG.debug("Unrolled steps: " + StepUtils.stepNamesAsString(steps)); this.setState(StepState.FINISHED); return steps; } LoopStep(String name); }
LoopStep extends RefactorStep implements ProvidesValidations { @Override public Set<Step> refactor(Set<Step> steps) { List<Row> iterationRows = getIterationRows(steps); String mode = getMode(); Set<Step> loopGraphSteps = StepUtils.getImmediateDependentSteps(this, steps); LOG.debug("Loop graph steps found: " + StepUtils.stepNamesAsString(loopGraphSteps)); Set<Step> loopGraphStepDependents = Sets.newHashSet(); for (Step loopGraphStep : loopGraphSteps) { for (Step loopGraphStepDependentCandidate : StepUtils.getImmediateDependentSteps(loopGraphStep, steps)) { if (!loopGraphSteps.contains(loopGraphStepDependentCandidate)) { loopGraphStepDependents.add(loopGraphStepDependentCandidate); } } } LOG.debug("Loop graph step dependents found: " + StepUtils.stepNamesAsString(loopGraphStepDependents)); steps.removeAll(loopGraphSteps); Set<Step> previousIterationSteps = null; for (Row iterationRow : iterationRows) { String suffix = getSuffix(iterationRow) ; LOG.debug("Constructing loop for iteration value: {}", suffix); Set<Step> iterationSteps = StepUtils.copySteps(loopGraphSteps); for (Step iterationStep : iterationSteps) { LOG.debug("Adjusting dependencies for iteration step: " + iterationStep.getName()); Set<String> dependencies = iterationStep.getDependencyNames(); Set<String> dependenciesToAdd = Sets.newHashSet(); Set<String> dependenciesToRemove = Sets.newHashSet(); for (String dependency : dependencies) { if (StepUtils.getStepForName(dependency, loopGraphSteps).isPresent()) { dependenciesToAdd.add(dependency + "_" + suffix); dependenciesToRemove.add(dependency); } } dependencies.addAll(dependenciesToAdd); LOG.debug("Added dependencies: " + dependenciesToAdd); dependencies.removeAll(dependenciesToRemove); LOG.debug("Removed dependencies: " + dependenciesToRemove); } for (Step iterationStep : iterationSteps) { String adjustedIterationStepName = iterationStep.getName() + "_" + suffix; LOG.debug("Renaming iteration step {} to {}", iterationStep.getName(), adjustedIterationStepName); iterationStep.setName(adjustedIterationStepName); for (Step loopGraphStepDependent : loopGraphStepDependents) { if (!(iterationStep instanceof LoopStep)) { LOG.debug("Adding dependency {} to {}", iterationStep.getName(), loopGraphStepDependent.getName()); loopGraphStepDependent.getDependencyNames().add(adjustedIterationStepName); } } Config parameterSubstConfig = performSubstitutions(iterationStep.getConfig(), iterationRow); iterationStep.setConfig(parameterSubstConfig); LOG.debug("Parameter substitutions completed: {} ", parameterSubstConfig.toString()); } if (mode.equals(MODE_SERIAL) && previousIterationSteps != null) { LOG.debug("Adjusting iteration steps to be dependent on previous iteration's steps"); for (Step iterationStep : iterationSteps) { for (Step previousIterationStep : previousIterationSteps) { if (!(previousIterationStep instanceof LoopStep)) { LOG.debug("Adding dependency {} to iteration step {}", previousIterationStep.getName(), iterationStep.getName()); iterationStep.getDependencyNames().add(previousIterationStep.getName()); } } } } previousIterationSteps = iterationSteps; steps.addAll(iterationSteps); } for (Step loopGraphStepDependent : loopGraphStepDependents) { for (Step loopGraphStep : loopGraphSteps) { loopGraphStepDependent.getDependencyNames().remove(loopGraphStep.getName()); } } LOG.debug("Unrolled steps: " + StepUtils.stepNamesAsString(steps)); this.setState(StepState.FINISHED); return steps; } LoopStep(String name); @Override Set<Step> refactor(Set<Step> steps); @Override Step copy(); @Override Validations getValidations(); }
LoopStep extends RefactorStep implements ProvidesValidations { @Override public Set<Step> refactor(Set<Step> steps) { List<Row> iterationRows = getIterationRows(steps); String mode = getMode(); Set<Step> loopGraphSteps = StepUtils.getImmediateDependentSteps(this, steps); LOG.debug("Loop graph steps found: " + StepUtils.stepNamesAsString(loopGraphSteps)); Set<Step> loopGraphStepDependents = Sets.newHashSet(); for (Step loopGraphStep : loopGraphSteps) { for (Step loopGraphStepDependentCandidate : StepUtils.getImmediateDependentSteps(loopGraphStep, steps)) { if (!loopGraphSteps.contains(loopGraphStepDependentCandidate)) { loopGraphStepDependents.add(loopGraphStepDependentCandidate); } } } LOG.debug("Loop graph step dependents found: " + StepUtils.stepNamesAsString(loopGraphStepDependents)); steps.removeAll(loopGraphSteps); Set<Step> previousIterationSteps = null; for (Row iterationRow : iterationRows) { String suffix = getSuffix(iterationRow) ; LOG.debug("Constructing loop for iteration value: {}", suffix); Set<Step> iterationSteps = StepUtils.copySteps(loopGraphSteps); for (Step iterationStep : iterationSteps) { LOG.debug("Adjusting dependencies for iteration step: " + iterationStep.getName()); Set<String> dependencies = iterationStep.getDependencyNames(); Set<String> dependenciesToAdd = Sets.newHashSet(); Set<String> dependenciesToRemove = Sets.newHashSet(); for (String dependency : dependencies) { if (StepUtils.getStepForName(dependency, loopGraphSteps).isPresent()) { dependenciesToAdd.add(dependency + "_" + suffix); dependenciesToRemove.add(dependency); } } dependencies.addAll(dependenciesToAdd); LOG.debug("Added dependencies: " + dependenciesToAdd); dependencies.removeAll(dependenciesToRemove); LOG.debug("Removed dependencies: " + dependenciesToRemove); } for (Step iterationStep : iterationSteps) { String adjustedIterationStepName = iterationStep.getName() + "_" + suffix; LOG.debug("Renaming iteration step {} to {}", iterationStep.getName(), adjustedIterationStepName); iterationStep.setName(adjustedIterationStepName); for (Step loopGraphStepDependent : loopGraphStepDependents) { if (!(iterationStep instanceof LoopStep)) { LOG.debug("Adding dependency {} to {}", iterationStep.getName(), loopGraphStepDependent.getName()); loopGraphStepDependent.getDependencyNames().add(adjustedIterationStepName); } } Config parameterSubstConfig = performSubstitutions(iterationStep.getConfig(), iterationRow); iterationStep.setConfig(parameterSubstConfig); LOG.debug("Parameter substitutions completed: {} ", parameterSubstConfig.toString()); } if (mode.equals(MODE_SERIAL) && previousIterationSteps != null) { LOG.debug("Adjusting iteration steps to be dependent on previous iteration's steps"); for (Step iterationStep : iterationSteps) { for (Step previousIterationStep : previousIterationSteps) { if (!(previousIterationStep instanceof LoopStep)) { LOG.debug("Adding dependency {} to iteration step {}", previousIterationStep.getName(), iterationStep.getName()); iterationStep.getDependencyNames().add(previousIterationStep.getName()); } } } } previousIterationSteps = iterationSteps; steps.addAll(iterationSteps); } for (Step loopGraphStepDependent : loopGraphStepDependents) { for (Step loopGraphStep : loopGraphSteps) { loopGraphStepDependent.getDependencyNames().remove(loopGraphStep.getName()); } } LOG.debug("Unrolled steps: " + StepUtils.stepNamesAsString(steps)); this.setState(StepState.FINISHED); return steps; } LoopStep(String name); @Override Set<Step> refactor(Set<Step> steps); @Override Step copy(); @Override Validations getValidations(); static final String MODE_PROPERTY; static final String MODE_SERIAL; static final String MODE_PARALLEL; static final String PARAMETER_PROPERTY; static final String SOURCE_PROPERTY; static final String SOURCE_RANGE; static final String SOURCE_LIST; static final String SOURCE_STEP; static final String RANGE_START_PROPERTY; static final String RANGE_END_PROPERTY; static final String LIST_PROPERTY; static final String STEP_PROPERTY; static final String SUFFIX_PROPERTY; static final String DEFAULT_SUFFIX_PROPERTY; }