PolicyImpl implements Policy { @Override public IRDAG runCompileTimeOptimization(final IRDAG dag, final String dagDirectory) { LOG.info("Launch Compile-time optimizations"); return process(dag, compileTimePasses.iterator(), dagDirectory); } PolicyImpl(final List<CompileTimePass> compileTimePasses, final Set<RunTimePass<?>> runTimePasses); @Override IRDAG runCompileTimeOptimization(final IRDAG dag, final String dagDirectory); @Override IRDAG runRunTimeOptimizations(final IRDAG irdag, final Message message);  }
@Test public void testTransientPolicy() throws Exception { final IRDAG optimizedDAG = TransientResourcePolicy.BUILDER.build().runCompileTimeOptimization(dag, DAG.EMPTY_DAG_DIRECTORY); assertTrue(optimizedDAG.checkIntegrity().isPassed()); }
@Test public void testDisaggregationPolicy() throws Exception { final IRDAG optimizedDAG = DisaggregationPolicy.BUILDER.build().runCompileTimeOptimization(dag, DAG.EMPTY_DAG_DIRECTORY); assertTrue(optimizedDAG.checkIntegrity().isPassed()); }
@Test public void testDataSkewPolicy() throws Exception { final IRDAG optimizedDAG = DataSkewPolicy.BUILDER.build().runCompileTimeOptimization(dagForSkew, DAG.EMPTY_DAG_DIRECTORY); assertTrue(optimizedDAG.checkIntegrity().isPassed()); }
@Test public void testLargeShufflePolicy() throws Exception { final IRDAG optimizedDAG = LargeShufflePolicy.BUILDER.build().runCompileTimeOptimization(dag, DAG.EMPTY_DAG_DIRECTORY); assertTrue(optimizedDAG.checkIntegrity().isPassed()); }
@Test public void testTransientAndLargeShuffleCombination() throws Exception { final List<CompileTimePass> compileTimePasses = new ArrayList<>(); final Set<RunTimePass<?>> runTimePasses = new HashSet<>(); compileTimePasses.addAll(LargeShufflePolicy.BUILDER.getCompileTimePasses()); runTimePasses.addAll(LargeShufflePolicy.BUILDER.getRunTimePasses()); compileTimePasses.addAll(TransientResourcePolicy.BUILDER.getCompileTimePasses()); runTimePasses.addAll(TransientResourcePolicy.BUILDER.getRunTimePasses()); final Policy combinedPolicy = new PolicyImpl(compileTimePasses, runTimePasses); final IRDAG optimizedDAG = combinedPolicy.runCompileTimeOptimization(dag, DAG.EMPTY_DAG_DIRECTORY); assertTrue(optimizedDAG.checkIntegrity().isPassed()); }
@Test public void testDataSkewAndLargeShuffleCombination() throws Exception { final List<CompileTimePass> compileTimePasses = new ArrayList<>(); final Set<RunTimePass<?>> runTimePasses = new HashSet<>(); compileTimePasses.addAll(DataSkewPolicy.BUILDER.getCompileTimePasses()); runTimePasses.addAll(DataSkewPolicy.BUILDER.getRunTimePasses()); compileTimePasses.addAll(LargeShufflePolicy.BUILDER.getCompileTimePasses()); runTimePasses.addAll(LargeShufflePolicy.BUILDER.getRunTimePasses()); final Policy combinedPolicy = new PolicyImpl(compileTimePasses, runTimePasses); expectedException.expect(CompileTimeOptimizationException.class); combinedPolicy.runCompileTimeOptimization(dagForSkew, DAG.EMPTY_DAG_DIRECTORY); }