@Test public void testForkAndMergeChaining() throws InjectionException { final MISTQueryBuilder queryBuilder = new MISTQueryBuilder(); final ContinuousStream<String> src1 = queryBuilder.socketTextStream(TestParameters.LOCAL_TEXT_SOCKET_SOURCE_CONF); final ContinuousStream<String> opA = src1.filter((x) -> true); final ContinuousStream<String> opB1 = opA.filter((x) -> true); final ContinuousStream<String> opB2 = opA.filter((x) -> true); final ContinuousStream<String> opB3 = opA.filter((x) -> true); final ContinuousStream<String> opC = opB2.union(opB1); final ContinuousStream<String> opD = opC.union(opB3); final MISTStream<String> sink1 = opD.textSocketOutput(TestParameters.HOST, TestParameters.SINK_PORT); queryBuilder.setApplicationId(TestParameters.SUPER_GROUP_ID); final MISTQuery query = queryBuilder.build(); final DAG<MISTStream, MISTEdge> dag = query.getDAG(); final OperatorChainDagGenerator chainDagGenerator = new OperatorChainDagGenerator(dag); final DAG<List<MISTStream>, MISTEdge> operatorChainDag = chainDagGenerator.generateOperatorChainDAG(); final List<MISTStream> src1List = Arrays.asList(src1); final List<MISTStream> opAList = Arrays.asList(opA); final List<MISTStream> opB1List = Arrays.asList(opB1); final List<MISTStream> opB2List = Arrays.asList(opB2); final List<MISTStream> opB3List = Arrays.asList(opB3); final List<MISTStream> opCList = Arrays.asList(opC); final List<MISTStream> opDList = Arrays.asList(opD); final List<MISTStream> sink1List = Arrays.asList(sink1); final Map<List<MISTStream>, MISTEdge> e1 = operatorChainDag.getEdges(src1List); final Map<List<MISTStream>, MISTEdge> result1 = new HashMap<>(); result1.put(opAList, new MISTEdge(Direction.LEFT)); Assert.assertEquals(result1, e1); final Map<List<MISTStream>, MISTEdge> e2 = operatorChainDag.getEdges(opAList); final Map<List<MISTStream>, MISTEdge> result2 = new HashMap<>(); result2.put(opB1List, new MISTEdge(Direction.LEFT)); result2.put(opB2List, new MISTEdge(Direction.LEFT)); result2.put(opB3List, new MISTEdge(Direction.LEFT)); Assert.assertEquals(result2, e2); final Map<List<MISTStream>, MISTEdge> e3 = operatorChainDag.getEdges(opB1List); final Map<List<MISTStream>, MISTEdge> result3 = new HashMap<>(); result3.put(opCList, new MISTEdge(Direction.RIGHT)); Assert.assertEquals(result3, e3); final Map<List<MISTStream>, MISTEdge> e4 = operatorChainDag.getEdges(opB2List); final Map<List<MISTStream>, MISTEdge> result4 = new HashMap<>(); result4.put(opCList, new MISTEdge(Direction.LEFT)); Assert.assertEquals(result4, e4); final Map<List<MISTStream>, MISTEdge> e5 = operatorChainDag.getEdges(opCList); final Map<List<MISTStream>, MISTEdge> result5 = new HashMap<>(); result5.put(opDList, new MISTEdge(Direction.LEFT)); Assert.assertEquals(result5, e5); final Map<List<MISTStream>, MISTEdge> e6 = operatorChainDag.getEdges(opB3List); final Map<List<MISTStream>, MISTEdge> result6 = new HashMap<>(); result6.put(opDList, new MISTEdge(Direction.RIGHT)); Assert.assertEquals(result6, e6); final Map<List<MISTStream>, MISTEdge> e7 = operatorChainDag.getEdges(opDList); final Map<List<MISTStream>, MISTEdge> result7 = new HashMap<>(); result7.put(sink1List, new MISTEdge(Direction.LEFT)); Assert.assertEquals(result7, e7); }
public DAG<List<MISTStream>, MISTEdge> generateOperatorChainDAG() { final DAG<OperatorChain, MISTEdge> chainDag = new AdjacentListDAG<>(); final Map<MISTStream, OperatorChain> vertexChainMap = new HashMap<>(); for (final MISTStream source : optimizedDag.getRootVertices()) { final Map<MISTStream, MISTEdge> rootEdges = optimizedDag.getEdges(source); final OperatorChain srcChain = new OperatorChain(); srcChain.chain.add(source); chainDag.addVertex(srcChain); vertexChainMap.put(source, srcChain); for (final Map.Entry<MISTStream, MISTEdge> entry : rootEdges.entrySet()) { final MISTStream nextVertex = entry.getKey(); final MISTEdge edge = entry.getValue(); final OperatorChain nextChain = getOperatorChain(nextVertex, vertexChainMap, chainDag); chainDag.addEdge(srcChain, nextChain, edge); chainingInDfsOrder(nextChain, nextVertex, chainDag, vertexChainMap); } } final DAG<List<MISTStream>, MISTEdge> result = new AdjacentListDAG<>(); final Queue<OperatorChain> queue = new LinkedList<>(); final Iterator<OperatorChain> iterator = GraphUtils.topologicalSort(chainDag); while (iterator.hasNext()) { final OperatorChain queryPartition = iterator.next(); queue.add(queryPartition); result.addVertex(queryPartition.chain); } for (final OperatorChain operatorChain : queue) { final Map<OperatorChain, MISTEdge> edges = chainDag.getEdges(operatorChain); for (final Map.Entry<OperatorChain, MISTEdge> edge : edges.entrySet()) { result.addEdge(operatorChain.chain, edge.getKey().chain, edge.getValue()); } } return result; }
OperatorChainDagGenerator { public DAG<List<MISTStream>, MISTEdge> generateOperatorChainDAG() { final DAG<OperatorChain, MISTEdge> chainDag = new AdjacentListDAG<>(); final Map<MISTStream, OperatorChain> vertexChainMap = new HashMap<>(); for (final MISTStream source : optimizedDag.getRootVertices()) { final Map<MISTStream, MISTEdge> rootEdges = optimizedDag.getEdges(source); final OperatorChain srcChain = new OperatorChain(); srcChain.chain.add(source); chainDag.addVertex(srcChain); vertexChainMap.put(source, srcChain); for (final Map.Entry<MISTStream, MISTEdge> entry : rootEdges.entrySet()) { final MISTStream nextVertex = entry.getKey(); final MISTEdge edge = entry.getValue(); final OperatorChain nextChain = getOperatorChain(nextVertex, vertexChainMap, chainDag); chainDag.addEdge(srcChain, nextChain, edge); chainingInDfsOrder(nextChain, nextVertex, chainDag, vertexChainMap); } } final DAG<List<MISTStream>, MISTEdge> result = new AdjacentListDAG<>(); final Queue<OperatorChain> queue = new LinkedList<>(); final Iterator<OperatorChain> iterator = GraphUtils.topologicalSort(chainDag); while (iterator.hasNext()) { final OperatorChain queryPartition = iterator.next(); queue.add(queryPartition); result.addVertex(queryPartition.chain); } for (final OperatorChain operatorChain : queue) { final Map<OperatorChain, MISTEdge> edges = chainDag.getEdges(operatorChain); for (final Map.Entry<OperatorChain, MISTEdge> edge : edges.entrySet()) { result.addEdge(operatorChain.chain, edge.getKey().chain, edge.getValue()); } } return result; } }
OperatorChainDagGenerator { public DAG<List<MISTStream>, MISTEdge> generateOperatorChainDAG() { final DAG<OperatorChain, MISTEdge> chainDag = new AdjacentListDAG<>(); final Map<MISTStream, OperatorChain> vertexChainMap = new HashMap<>(); for (final MISTStream source : optimizedDag.getRootVertices()) { final Map<MISTStream, MISTEdge> rootEdges = optimizedDag.getEdges(source); final OperatorChain srcChain = new OperatorChain(); srcChain.chain.add(source); chainDag.addVertex(srcChain); vertexChainMap.put(source, srcChain); for (final Map.Entry<MISTStream, MISTEdge> entry : rootEdges.entrySet()) { final MISTStream nextVertex = entry.getKey(); final MISTEdge edge = entry.getValue(); final OperatorChain nextChain = getOperatorChain(nextVertex, vertexChainMap, chainDag); chainDag.addEdge(srcChain, nextChain, edge); chainingInDfsOrder(nextChain, nextVertex, chainDag, vertexChainMap); } } final DAG<List<MISTStream>, MISTEdge> result = new AdjacentListDAG<>(); final Queue<OperatorChain> queue = new LinkedList<>(); final Iterator<OperatorChain> iterator = GraphUtils.topologicalSort(chainDag); while (iterator.hasNext()) { final OperatorChain queryPartition = iterator.next(); queue.add(queryPartition); result.addVertex(queryPartition.chain); } for (final OperatorChain operatorChain : queue) { final Map<OperatorChain, MISTEdge> edges = chainDag.getEdges(operatorChain); for (final Map.Entry<OperatorChain, MISTEdge> edge : edges.entrySet()) { result.addEdge(operatorChain.chain, edge.getKey().chain, edge.getValue()); } } return result; } OperatorChainDagGenerator(final DAG<MISTStream, MISTEdge> optimizedDag); }
OperatorChainDagGenerator { public DAG<List<MISTStream>, MISTEdge> generateOperatorChainDAG() { final DAG<OperatorChain, MISTEdge> chainDag = new AdjacentListDAG<>(); final Map<MISTStream, OperatorChain> vertexChainMap = new HashMap<>(); for (final MISTStream source : optimizedDag.getRootVertices()) { final Map<MISTStream, MISTEdge> rootEdges = optimizedDag.getEdges(source); final OperatorChain srcChain = new OperatorChain(); srcChain.chain.add(source); chainDag.addVertex(srcChain); vertexChainMap.put(source, srcChain); for (final Map.Entry<MISTStream, MISTEdge> entry : rootEdges.entrySet()) { final MISTStream nextVertex = entry.getKey(); final MISTEdge edge = entry.getValue(); final OperatorChain nextChain = getOperatorChain(nextVertex, vertexChainMap, chainDag); chainDag.addEdge(srcChain, nextChain, edge); chainingInDfsOrder(nextChain, nextVertex, chainDag, vertexChainMap); } } final DAG<List<MISTStream>, MISTEdge> result = new AdjacentListDAG<>(); final Queue<OperatorChain> queue = new LinkedList<>(); final Iterator<OperatorChain> iterator = GraphUtils.topologicalSort(chainDag); while (iterator.hasNext()) { final OperatorChain queryPartition = iterator.next(); queue.add(queryPartition); result.addVertex(queryPartition.chain); } for (final OperatorChain operatorChain : queue) { final Map<OperatorChain, MISTEdge> edges = chainDag.getEdges(operatorChain); for (final Map.Entry<OperatorChain, MISTEdge> edge : edges.entrySet()) { result.addEdge(operatorChain.chain, edge.getKey().chain, edge.getValue()); } } return result; } OperatorChainDagGenerator(final DAG<MISTStream, MISTEdge> optimizedDag); DAG<List<MISTStream>, MISTEdge> generateOperatorChainDAG(); }
OperatorChainDagGenerator { public DAG<List<MISTStream>, MISTEdge> generateOperatorChainDAG() { final DAG<OperatorChain, MISTEdge> chainDag = new AdjacentListDAG<>(); final Map<MISTStream, OperatorChain> vertexChainMap = new HashMap<>(); for (final MISTStream source : optimizedDag.getRootVertices()) { final Map<MISTStream, MISTEdge> rootEdges = optimizedDag.getEdges(source); final OperatorChain srcChain = new OperatorChain(); srcChain.chain.add(source); chainDag.addVertex(srcChain); vertexChainMap.put(source, srcChain); for (final Map.Entry<MISTStream, MISTEdge> entry : rootEdges.entrySet()) { final MISTStream nextVertex = entry.getKey(); final MISTEdge edge = entry.getValue(); final OperatorChain nextChain = getOperatorChain(nextVertex, vertexChainMap, chainDag); chainDag.addEdge(srcChain, nextChain, edge); chainingInDfsOrder(nextChain, nextVertex, chainDag, vertexChainMap); } } final DAG<List<MISTStream>, MISTEdge> result = new AdjacentListDAG<>(); final Queue<OperatorChain> queue = new LinkedList<>(); final Iterator<OperatorChain> iterator = GraphUtils.topologicalSort(chainDag); while (iterator.hasNext()) { final OperatorChain queryPartition = iterator.next(); queue.add(queryPartition); result.addVertex(queryPartition.chain); } for (final OperatorChain operatorChain : queue) { final Map<OperatorChain, MISTEdge> edges = chainDag.getEdges(operatorChain); for (final Map.Entry<OperatorChain, MISTEdge> edge : edges.entrySet()) { result.addEdge(operatorChain.chain, edge.getKey().chain, edge.getValue()); } } return result; } OperatorChainDagGenerator(final DAG<MISTStream, MISTEdge> optimizedDag); DAG<List<MISTStream>, MISTEdge> generateOperatorChainDAG();  }