@Test public void testSequentialChaining() throws InjectionException { final MISTQueryBuilder queryBuilder = new MISTQueryBuilder(); final ContinuousStream<String> src1 = 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> op13 = op12.filter((x) -> true); final MISTStream<String> sink1 = op13.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> opList = Arrays.asList(op11, op12, op13); final List<MISTStream> sinkList = Arrays.asList(sink1); final Map<List<MISTStream>, MISTEdge> e1 = operatorChainDag.getEdges(src1List); final Map<List<MISTStream>, MISTEdge> result1 = new HashMap<>(); result1.put(opList, new MISTEdge(Direction.LEFT)); Assert.assertEquals(e1, result1); final Map<List<MISTStream>, MISTEdge> e2 = operatorChainDag.getEdges(opList); final Map<List<MISTStream>, MISTEdge> result2 = new HashMap<>(); result2.put(sinkList, new MISTEdge(Direction.LEFT)); Assert.assertEquals(e2, result2); }
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();  }