@Test public void testMergingQueryPartitioning() throws InjectionException { final MISTQueryBuilder queryBuilder = new MISTQueryBuilder(); final ContinuousStream<String> src1 = queryBuilder.socketTextStream(TestParameters.LOCAL_TEXT_SOCKET_SOURCE_CONF); final ContinuousStream<String> src2 = queryBuilder.socketTextStream(TestParameters.LOCAL_TEXT_SOCKET_SOURCE_CONF); final ContinuousStream<String> src3 = queryBuilder.socketTextStream(TestParameters.LOCAL_TEXT_SOCKET_SOURCE_CONF); final ContinuousStream<String> op11 = src1.filter((x) -> true); final ContinuousStream<String> op12 = op11.filter((x) -> true); final ContinuousStream<String> op21 = src2.filter((x) -> true); final ContinuousStream<String> op13 = op12.union(op21); final ContinuousStream<String> op31 = src3.filter((x) -> true); final ContinuousStream<String> op14 = op13.union(op31); final MISTStream<String> sink1 = op14.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> src2List = Arrays.asList(src2); final List<MISTStream> src3List = Arrays.asList(src3); final List<MISTStream> op11op12 = Arrays.asList(op11, op12); final List<MISTStream> op21List = Arrays.asList(op21); final List<MISTStream> op13List = Arrays.asList(op13); final List<MISTStream> op31List = Arrays.asList(op31); final List<MISTStream> op14List = Arrays.asList(op14); 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(op11op12, new MISTEdge(Direction.LEFT)); Assert.assertEquals(e1, result1); final Map<List<MISTStream>, MISTEdge> e2 = operatorChainDag.getEdges(src2List); final Map<List<MISTStream>, MISTEdge> result2 = new HashMap<>(); result2.put(op21List, new MISTEdge(Direction.LEFT)); Assert.assertEquals(e2, result2); final Map<List<MISTStream>, MISTEdge> e3 = operatorChainDag.getEdges(src3List); final Map<List<MISTStream>, MISTEdge> result3 = new HashMap<>(); result3.put(op31List, new MISTEdge(Direction.LEFT)); Assert.assertEquals(e3, result3); final Map<List<MISTStream>, MISTEdge> e4 = operatorChainDag.getEdges(op11op12); final Map<List<MISTStream>, MISTEdge> result4 = new HashMap<>(); result4.put(op13List, new MISTEdge(Direction.LEFT)); Assert.assertEquals(e4, result4); final Map<List<MISTStream>, MISTEdge> e5 = operatorChainDag.getEdges(op21List); final Map<List<MISTStream>, MISTEdge> result5 = new HashMap<>(); result5.put(op13List, new MISTEdge(Direction.RIGHT)); Assert.assertEquals(e5, result5); final Map<List<MISTStream>, MISTEdge> e6 = operatorChainDag.getEdges(op13List); final Map<List<MISTStream>, MISTEdge> result6 = new HashMap<>(); result6.put(op14List, new MISTEdge(Direction.LEFT)); Assert.assertEquals(e6, result6); final Map<List<MISTStream>, MISTEdge> e7 = operatorChainDag.getEdges(op31List); final Map<List<MISTStream>, MISTEdge> result7 = new HashMap<>(); result7.put(op14List, new MISTEdge(Direction.RIGHT)); Assert.assertEquals(e7, result7); final Map<List<MISTStream>, MISTEdge> e8 = operatorChainDag.getEdges(op14List); final Map<List<MISTStream>, MISTEdge> result8 = new HashMap<>(); result8.put(sink1List, new MISTEdge(Direction.LEFT)); Assert.assertEquals(e8, result8); }
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();  }