@Test public void mergingDifferentSourceQueriesOneGroupTest() throws InjectionException, IOException, ClassNotFoundException { final List<String> result1 = new LinkedList<>(); final List<String> paths1 = mock(List.class); final Map<String, String> sourceConf1 = idAndConfGenerator.generateConf(); final Map<String, String> ocConf1 = idAndConfGenerator.generateConf(); final Map<String, String> sinkConf1 = idAndConfGenerator.generateConf(); final TestSource src1 = generateSource(sourceConf1); final PhysicalOperator physicalOp1 = generateFilterOperator(ocConf1, (s) -> true); final PhysicalSink<String> sink1 = generateSink(sinkConf1, result1); final ConfigVertex srcVertex1 = new ConfigVertex(Long.toString(configVertexId.getAndIncrement()), ExecutionVertex.Type.SOURCE, sourceConf1); final ConfigVertex ocVertex1 = new ConfigVertex(Long.toString(configVertexId.getAndIncrement()), ExecutionVertex.Type.OPERATOR, ocConf1); final ConfigVertex sinkVertex1 = new ConfigVertex(Long.toString(configVertexId.getAndIncrement()), ExecutionVertex.Type.SINK, sinkConf1); final Tuple<DAG<ConfigVertex, MISTEdge>, ExecutionDag> dagTuple1 = generateSimpleDag(src1, physicalOp1, sink1, srcVertex1, ocVertex1, sinkVertex1); final List<String> result2 = new LinkedList<>(); final List<String> paths2 = mock(List.class); final Map<String, String> sourceConf2 = idAndConfGenerator.generateConf(); final Map<String, String> ocConf2 = idAndConfGenerator.generateConf(); final Map<String, String> sinkConf2 = idAndConfGenerator.generateConf(); final TestSource src2 = generateSource(sourceConf2); final PhysicalOperator physicalOp2 = generateFilterOperator(ocConf2, (s) -> true); final PhysicalSink<String> sink2 = generateSink(sinkConf2, result2); final ConfigVertex srcVertex2 = new ConfigVertex(Long.toString(configVertexId.getAndIncrement()), ExecutionVertex.Type.SOURCE, sourceConf2); final ConfigVertex ocVertex2 = new ConfigVertex(Long.toString(configVertexId.getAndIncrement()), ExecutionVertex.Type.OPERATOR, ocConf2); final ConfigVertex sinkVertex2 = new ConfigVertex(Long.toString(configVertexId.getAndIncrement()), ExecutionVertex.Type.SINK, sinkConf2); final Tuple<DAG<ConfigVertex, MISTEdge>, ExecutionDag> dagTuple2 = generateSimpleDag(src2, physicalOp2, sink2, srcVertex2, ocVertex2, sinkVertex2); final Query query1 = mock(Query.class); final Query query2 = mock(Query.class); final String query1Id = "q1"; final String query2Id = "q2"; queryStarter.start(query1Id, query1, dagTuple1.getKey(), paths1); queryStarter.start(query2Id, query2, dagTuple2.getKey(), paths2); final String data1 = "Hello"; src1.send(data1); Assert.assertEquals(1, src1.getSourceOutputEmitter().processAllEvent()); Assert.assertEquals(Arrays.asList(data1), result1); Assert.assertEquals(0, src2.getSourceOutputEmitter().processAllEvent()); Assert.assertEquals(Arrays.asList(), result2); final String data2 = "World"; src2.send(data2); Assert.assertEquals(1, src2.getSourceOutputEmitter().processAllEvent()); Assert.assertEquals(Arrays.asList(data2), result2); Assert.assertEquals(0, src1.getSourceOutputEmitter().processAllEvent()); Assert.assertEquals(Arrays.asList(data1), result1); Assert.assertEquals(dagTuple1.getKey(), queryIdConfigDagMap.get(query1Id)); Assert.assertEquals(dagTuple2.getKey(), queryIdConfigDagMap.get(query2Id)); final DAG<ExecutionVertex, MISTEdge> dag1 = dagTuple1.getValue().getDag(); Assert.assertTrue(GraphUtils.compareTwoDag(dag1, srcAndDagMap.get(sourceConf1).getDag())); final DAG<ExecutionVertex, MISTEdge> dag2 = dagTuple2.getValue().getDag(); Assert.assertTrue(GraphUtils.compareTwoDag(dag2, srcAndDagMap.get(sourceConf2).getDag())); Assert.assertTrue(GraphUtils.compareTwoDag(dag1, executionVertexDagMap.get(src1).getDag())); Assert.assertTrue(GraphUtils.compareTwoDag(dag1, executionVertexDagMap.get(physicalOp1).getDag())); Assert.assertTrue(GraphUtils.compareTwoDag(dag1, executionVertexDagMap.get(sink1).getDag())); Assert.assertTrue(GraphUtils.compareTwoDag(dag2, executionVertexDagMap.get(src2).getDag())); Assert.assertTrue(GraphUtils.compareTwoDag(dag2, executionVertexDagMap.get(physicalOp2).getDag())); Assert.assertTrue(GraphUtils.compareTwoDag(dag2, executionVertexDagMap.get(sink2).getDag())); Assert.assertEquals(src1, configExecutionVertexMap.get(srcVertex1)); Assert.assertEquals(physicalOp1, configExecutionVertexMap.get(ocVertex1)); Assert.assertEquals(sink1, configExecutionVertexMap.get(sinkVertex1)); Assert.assertEquals(src2, configExecutionVertexMap.get(srcVertex2)); Assert.assertEquals(physicalOp2, configExecutionVertexMap.get(ocVertex2)); Assert.assertEquals(sink2, configExecutionVertexMap.get(sinkVertex2)); Assert.assertEquals(1, (int)executionVertexCountMap.get(src1)); Assert.assertEquals(1, (int)executionVertexCountMap.get(physicalOp1)); Assert.assertEquals(1, (int)executionVertexCountMap.get(sink1)); Assert.assertEquals(1, (int)executionVertexCountMap.get(src2)); Assert.assertEquals(1, (int)executionVertexCountMap.get(physicalOp2)); Assert.assertEquals(1, (int)executionVertexCountMap.get(sink2)); final Collection<ExecutionDag> expectedDags = new HashSet<>(); expectedDags.add(srcAndDagMap.get(sourceConf1)); expectedDags.add(srcAndDagMap.get(sourceConf2)); Assert.assertEquals(expectedDags, executionDags.values()); }
@Override public synchronized void start(final String queryId, final Query query, final DAG<ConfigVertex, MISTEdge> submittedDag, final List<String> jarFilePaths) throws IOException, ClassNotFoundException { queryIdConfigDagMap.put(queryId, submittedDag); final URL[] urls = SerializeUtils.getJarFileURLs(jarFilePaths); final ClassLoader classLoader = classLoaderProvider.newInstance(urls); synchronized (groupJarFilePaths) { if (jarFilePaths != null && jarFilePaths.size() != 0) { groupJarFilePaths.addAll(jarFilePaths); } } synchronized (srcAndDagMap) { final Map<Map<String, String>, ExecutionDag> mergeableDags = findMergeableDags(submittedDag); if (mergeableDags.size() == 0) { final ExecutionDag executionDag = generate(submittedDag, urls, classLoader); QueryStarterUtils.setUpOutputEmitters(executionDag, query); for (final ExecutionVertex source : executionDag.getDag().getRootVertices()) { final PhysicalSource src = (PhysicalSource) source; srcAndDagMap.put(src.getConfiguration(), executionDag); src.start(); } for (final ExecutionVertex ev : executionDag.getDag().getVertices()) { executionVertexDagMap.put(ev, executionDag); } executionDags.add(executionDag); return; } final ExecutionDag sharableExecutionDag = selectLargestDag(mergeableDags.values()); for (final ExecutionDag executionDag : mergeableDags.values()) { if (executionDag != sharableExecutionDag) { GraphUtils.copy(executionDag.getDag(), sharableExecutionDag.getDag()); executionDags.remove(executionDag); for (final ExecutionVertex source : executionDag.getDag().getRootVertices()) { srcAndDagMap.replace(((PhysicalSource) source).getConfiguration(), sharableExecutionDag); } for (final ExecutionVertex ev : executionDag.getDag().getVertices()) { executionVertexDagMap.put(ev, sharableExecutionDag); } } } final Map<ConfigVertex, ExecutionVertex> subDagMap = commonSubDagFinder.findSubDag(sharableExecutionDag, submittedDag); final Set<ConfigVertex> visited = new HashSet<>(submittedDag.numberOfVertices()); for (final ConfigVertex source : submittedDag.getRootVertices()) { ExecutionVertex executionVertex; if (subDagMap.get(source) == null) { executionVertex = executionVertexGenerator.generate(source, urls, classLoader); sharableExecutionDag.getDag().addVertex(executionVertex); executionVertexCountMap.put(executionVertex, 1); executionVertexDagMap.put(executionVertex, sharableExecutionDag); } else { executionVertex = subDagMap.get(source); executionVertexCountMap.put(executionVertex, executionVertexCountMap.get(executionVertex) + 1); } configExecutionVertexMap.put(source, executionVertex); for (final Map.Entry<ConfigVertex, MISTEdge> child : submittedDag.getEdges(source).entrySet()) { dfsMerge(subDagMap, visited, executionVertex, child.getValue(), child.getKey(), sharableExecutionDag, submittedDag, urls, classLoader); } } for (final ConfigVertex source : submittedDag.getRootVertices()) { if (!subDagMap.containsKey(source)) { srcAndDagMap.put(source.getConfiguration(), sharableExecutionDag); ((PhysicalSource)configExecutionVertexMap.get(source)).start(); } } } }
ImmediateQueryMergingStarter implements QueryStarter { @Override public synchronized void start(final String queryId, final Query query, final DAG<ConfigVertex, MISTEdge> submittedDag, final List<String> jarFilePaths) throws IOException, ClassNotFoundException { queryIdConfigDagMap.put(queryId, submittedDag); final URL[] urls = SerializeUtils.getJarFileURLs(jarFilePaths); final ClassLoader classLoader = classLoaderProvider.newInstance(urls); synchronized (groupJarFilePaths) { if (jarFilePaths != null && jarFilePaths.size() != 0) { groupJarFilePaths.addAll(jarFilePaths); } } synchronized (srcAndDagMap) { final Map<Map<String, String>, ExecutionDag> mergeableDags = findMergeableDags(submittedDag); if (mergeableDags.size() == 0) { final ExecutionDag executionDag = generate(submittedDag, urls, classLoader); QueryStarterUtils.setUpOutputEmitters(executionDag, query); for (final ExecutionVertex source : executionDag.getDag().getRootVertices()) { final PhysicalSource src = (PhysicalSource) source; srcAndDagMap.put(src.getConfiguration(), executionDag); src.start(); } for (final ExecutionVertex ev : executionDag.getDag().getVertices()) { executionVertexDagMap.put(ev, executionDag); } executionDags.add(executionDag); return; } final ExecutionDag sharableExecutionDag = selectLargestDag(mergeableDags.values()); for (final ExecutionDag executionDag : mergeableDags.values()) { if (executionDag != sharableExecutionDag) { GraphUtils.copy(executionDag.getDag(), sharableExecutionDag.getDag()); executionDags.remove(executionDag); for (final ExecutionVertex source : executionDag.getDag().getRootVertices()) { srcAndDagMap.replace(((PhysicalSource) source).getConfiguration(), sharableExecutionDag); } for (final ExecutionVertex ev : executionDag.getDag().getVertices()) { executionVertexDagMap.put(ev, sharableExecutionDag); } } } final Map<ConfigVertex, ExecutionVertex> subDagMap = commonSubDagFinder.findSubDag(sharableExecutionDag, submittedDag); final Set<ConfigVertex> visited = new HashSet<>(submittedDag.numberOfVertices()); for (final ConfigVertex source : submittedDag.getRootVertices()) { ExecutionVertex executionVertex; if (subDagMap.get(source) == null) { executionVertex = executionVertexGenerator.generate(source, urls, classLoader); sharableExecutionDag.getDag().addVertex(executionVertex); executionVertexCountMap.put(executionVertex, 1); executionVertexDagMap.put(executionVertex, sharableExecutionDag); } else { executionVertex = subDagMap.get(source); executionVertexCountMap.put(executionVertex, executionVertexCountMap.get(executionVertex) + 1); } configExecutionVertexMap.put(source, executionVertex); for (final Map.Entry<ConfigVertex, MISTEdge> child : submittedDag.getEdges(source).entrySet()) { dfsMerge(subDagMap, visited, executionVertex, child.getValue(), child.getKey(), sharableExecutionDag, submittedDag, urls, classLoader); } } for (final ConfigVertex source : submittedDag.getRootVertices()) { if (!subDagMap.containsKey(source)) { srcAndDagMap.put(source.getConfiguration(), sharableExecutionDag); ((PhysicalSource)configExecutionVertexMap.get(source)).start(); } } } } }
ImmediateQueryMergingStarter implements QueryStarter { @Override public synchronized void start(final String queryId, final Query query, final DAG<ConfigVertex, MISTEdge> submittedDag, final List<String> jarFilePaths) throws IOException, ClassNotFoundException { queryIdConfigDagMap.put(queryId, submittedDag); final URL[] urls = SerializeUtils.getJarFileURLs(jarFilePaths); final ClassLoader classLoader = classLoaderProvider.newInstance(urls); synchronized (groupJarFilePaths) { if (jarFilePaths != null && jarFilePaths.size() != 0) { groupJarFilePaths.addAll(jarFilePaths); } } synchronized (srcAndDagMap) { final Map<Map<String, String>, ExecutionDag> mergeableDags = findMergeableDags(submittedDag); if (mergeableDags.size() == 0) { final ExecutionDag executionDag = generate(submittedDag, urls, classLoader); QueryStarterUtils.setUpOutputEmitters(executionDag, query); for (final ExecutionVertex source : executionDag.getDag().getRootVertices()) { final PhysicalSource src = (PhysicalSource) source; srcAndDagMap.put(src.getConfiguration(), executionDag); src.start(); } for (final ExecutionVertex ev : executionDag.getDag().getVertices()) { executionVertexDagMap.put(ev, executionDag); } executionDags.add(executionDag); return; } final ExecutionDag sharableExecutionDag = selectLargestDag(mergeableDags.values()); for (final ExecutionDag executionDag : mergeableDags.values()) { if (executionDag != sharableExecutionDag) { GraphUtils.copy(executionDag.getDag(), sharableExecutionDag.getDag()); executionDags.remove(executionDag); for (final ExecutionVertex source : executionDag.getDag().getRootVertices()) { srcAndDagMap.replace(((PhysicalSource) source).getConfiguration(), sharableExecutionDag); } for (final ExecutionVertex ev : executionDag.getDag().getVertices()) { executionVertexDagMap.put(ev, sharableExecutionDag); } } } final Map<ConfigVertex, ExecutionVertex> subDagMap = commonSubDagFinder.findSubDag(sharableExecutionDag, submittedDag); final Set<ConfigVertex> visited = new HashSet<>(submittedDag.numberOfVertices()); for (final ConfigVertex source : submittedDag.getRootVertices()) { ExecutionVertex executionVertex; if (subDagMap.get(source) == null) { executionVertex = executionVertexGenerator.generate(source, urls, classLoader); sharableExecutionDag.getDag().addVertex(executionVertex); executionVertexCountMap.put(executionVertex, 1); executionVertexDagMap.put(executionVertex, sharableExecutionDag); } else { executionVertex = subDagMap.get(source); executionVertexCountMap.put(executionVertex, executionVertexCountMap.get(executionVertex) + 1); } configExecutionVertexMap.put(source, executionVertex); for (final Map.Entry<ConfigVertex, MISTEdge> child : submittedDag.getEdges(source).entrySet()) { dfsMerge(subDagMap, visited, executionVertex, child.getValue(), child.getKey(), sharableExecutionDag, submittedDag, urls, classLoader); } } for (final ConfigVertex source : submittedDag.getRootVertices()) { if (!subDagMap.containsKey(source)) { srcAndDagMap.put(source.getConfiguration(), sharableExecutionDag); ((PhysicalSource)configExecutionVertexMap.get(source)).start(); } } } } @Inject private  ImmediateQueryMergingStarter(final CommonSubDagFinder commonSubDagFinder,
                                       final SrcAndDagMap<Map<String, String>> srcAndDagMap,
                                       final QueryIdConfigDagMap queryIdConfigDagMap,
                                       final ExecutionDags executionDags,
                                       final ConfigExecutionVertexMap configExecutionVertexMap,
                                       final ExecutionVertexCountMap executionVertexCountMap,
                                       final ClassLoaderProvider classLoaderProvider,
                                       final ExecutionVertexGenerator executionVertexGenerator,
                                       final ExecutionVertexDagMap executionVertexDagMap); }
ImmediateQueryMergingStarter implements QueryStarter { @Override public synchronized void start(final String queryId, final Query query, final DAG<ConfigVertex, MISTEdge> submittedDag, final List<String> jarFilePaths) throws IOException, ClassNotFoundException { queryIdConfigDagMap.put(queryId, submittedDag); final URL[] urls = SerializeUtils.getJarFileURLs(jarFilePaths); final ClassLoader classLoader = classLoaderProvider.newInstance(urls); synchronized (groupJarFilePaths) { if (jarFilePaths != null && jarFilePaths.size() != 0) { groupJarFilePaths.addAll(jarFilePaths); } } synchronized (srcAndDagMap) { final Map<Map<String, String>, ExecutionDag> mergeableDags = findMergeableDags(submittedDag); if (mergeableDags.size() == 0) { final ExecutionDag executionDag = generate(submittedDag, urls, classLoader); QueryStarterUtils.setUpOutputEmitters(executionDag, query); for (final ExecutionVertex source : executionDag.getDag().getRootVertices()) { final PhysicalSource src = (PhysicalSource) source; srcAndDagMap.put(src.getConfiguration(), executionDag); src.start(); } for (final ExecutionVertex ev : executionDag.getDag().getVertices()) { executionVertexDagMap.put(ev, executionDag); } executionDags.add(executionDag); return; } final ExecutionDag sharableExecutionDag = selectLargestDag(mergeableDags.values()); for (final ExecutionDag executionDag : mergeableDags.values()) { if (executionDag != sharableExecutionDag) { GraphUtils.copy(executionDag.getDag(), sharableExecutionDag.getDag()); executionDags.remove(executionDag); for (final ExecutionVertex source : executionDag.getDag().getRootVertices()) { srcAndDagMap.replace(((PhysicalSource) source).getConfiguration(), sharableExecutionDag); } for (final ExecutionVertex ev : executionDag.getDag().getVertices()) { executionVertexDagMap.put(ev, sharableExecutionDag); } } } final Map<ConfigVertex, ExecutionVertex> subDagMap = commonSubDagFinder.findSubDag(sharableExecutionDag, submittedDag); final Set<ConfigVertex> visited = new HashSet<>(submittedDag.numberOfVertices()); for (final ConfigVertex source : submittedDag.getRootVertices()) { ExecutionVertex executionVertex; if (subDagMap.get(source) == null) { executionVertex = executionVertexGenerator.generate(source, urls, classLoader); sharableExecutionDag.getDag().addVertex(executionVertex); executionVertexCountMap.put(executionVertex, 1); executionVertexDagMap.put(executionVertex, sharableExecutionDag); } else { executionVertex = subDagMap.get(source); executionVertexCountMap.put(executionVertex, executionVertexCountMap.get(executionVertex) + 1); } configExecutionVertexMap.put(source, executionVertex); for (final Map.Entry<ConfigVertex, MISTEdge> child : submittedDag.getEdges(source).entrySet()) { dfsMerge(subDagMap, visited, executionVertex, child.getValue(), child.getKey(), sharableExecutionDag, submittedDag, urls, classLoader); } } for (final ConfigVertex source : submittedDag.getRootVertices()) { if (!subDagMap.containsKey(source)) { srcAndDagMap.put(source.getConfiguration(), sharableExecutionDag); ((PhysicalSource)configExecutionVertexMap.get(source)).start(); } } } } @Inject private  ImmediateQueryMergingStarter(final CommonSubDagFinder commonSubDagFinder,
                                       final SrcAndDagMap<Map<String, String>> srcAndDagMap,
                                       final QueryIdConfigDagMap queryIdConfigDagMap,
                                       final ExecutionDags executionDags,
                                       final ConfigExecutionVertexMap configExecutionVertexMap,
                                       final ExecutionVertexCountMap executionVertexCountMap,
                                       final ClassLoaderProvider classLoaderProvider,
                                       final ExecutionVertexGenerator executionVertexGenerator,
                                       final ExecutionVertexDagMap executionVertexDagMap); @Override synchronized void start(final String queryId,
                                 final Query query,
                                 final DAG<ConfigVertex, MISTEdge> submittedDag,
                                 final List<String> jarFilePaths); }
ImmediateQueryMergingStarter implements QueryStarter { @Override public synchronized void start(final String queryId, final Query query, final DAG<ConfigVertex, MISTEdge> submittedDag, final List<String> jarFilePaths) throws IOException, ClassNotFoundException { queryIdConfigDagMap.put(queryId, submittedDag); final URL[] urls = SerializeUtils.getJarFileURLs(jarFilePaths); final ClassLoader classLoader = classLoaderProvider.newInstance(urls); synchronized (groupJarFilePaths) { if (jarFilePaths != null && jarFilePaths.size() != 0) { groupJarFilePaths.addAll(jarFilePaths); } } synchronized (srcAndDagMap) { final Map<Map<String, String>, ExecutionDag> mergeableDags = findMergeableDags(submittedDag); if (mergeableDags.size() == 0) { final ExecutionDag executionDag = generate(submittedDag, urls, classLoader); QueryStarterUtils.setUpOutputEmitters(executionDag, query); for (final ExecutionVertex source : executionDag.getDag().getRootVertices()) { final PhysicalSource src = (PhysicalSource) source; srcAndDagMap.put(src.getConfiguration(), executionDag); src.start(); } for (final ExecutionVertex ev : executionDag.getDag().getVertices()) { executionVertexDagMap.put(ev, executionDag); } executionDags.add(executionDag); return; } final ExecutionDag sharableExecutionDag = selectLargestDag(mergeableDags.values()); for (final ExecutionDag executionDag : mergeableDags.values()) { if (executionDag != sharableExecutionDag) { GraphUtils.copy(executionDag.getDag(), sharableExecutionDag.getDag()); executionDags.remove(executionDag); for (final ExecutionVertex source : executionDag.getDag().getRootVertices()) { srcAndDagMap.replace(((PhysicalSource) source).getConfiguration(), sharableExecutionDag); } for (final ExecutionVertex ev : executionDag.getDag().getVertices()) { executionVertexDagMap.put(ev, sharableExecutionDag); } } } final Map<ConfigVertex, ExecutionVertex> subDagMap = commonSubDagFinder.findSubDag(sharableExecutionDag, submittedDag); final Set<ConfigVertex> visited = new HashSet<>(submittedDag.numberOfVertices()); for (final ConfigVertex source : submittedDag.getRootVertices()) { ExecutionVertex executionVertex; if (subDagMap.get(source) == null) { executionVertex = executionVertexGenerator.generate(source, urls, classLoader); sharableExecutionDag.getDag().addVertex(executionVertex); executionVertexCountMap.put(executionVertex, 1); executionVertexDagMap.put(executionVertex, sharableExecutionDag); } else { executionVertex = subDagMap.get(source); executionVertexCountMap.put(executionVertex, executionVertexCountMap.get(executionVertex) + 1); } configExecutionVertexMap.put(source, executionVertex); for (final Map.Entry<ConfigVertex, MISTEdge> child : submittedDag.getEdges(source).entrySet()) { dfsMerge(subDagMap, visited, executionVertex, child.getValue(), child.getKey(), sharableExecutionDag, submittedDag, urls, classLoader); } } for (final ConfigVertex source : submittedDag.getRootVertices()) { if (!subDagMap.containsKey(source)) { srcAndDagMap.put(source.getConfiguration(), sharableExecutionDag); ((PhysicalSource)configExecutionVertexMap.get(source)).start(); } } } } @Inject private  ImmediateQueryMergingStarter(final CommonSubDagFinder commonSubDagFinder,
                                       final SrcAndDagMap<Map<String, String>> srcAndDagMap,
                                       final QueryIdConfigDagMap queryIdConfigDagMap,
                                       final ExecutionDags executionDags,
                                       final ConfigExecutionVertexMap configExecutionVertexMap,
                                       final ExecutionVertexCountMap executionVertexCountMap,
                                       final ClassLoaderProvider classLoaderProvider,
                                       final ExecutionVertexGenerator executionVertexGenerator,
                                       final ExecutionVertexDagMap executionVertexDagMap); @Override synchronized void start(final String queryId,
                                 final Query query,
                                 final DAG<ConfigVertex, MISTEdge> submittedDag,
                                 final List<String> jarFilePaths);  }