@Test public void singleQueryMergingTest() throws InjectionException, IOException, ClassNotFoundException { final List<String> result = new LinkedList<>(); final Map<String, String> sourceConf = idAndConfGenerator.generateConf(); final Map<String, String> ocConf = idAndConfGenerator.generateConf(); final Map<String, String> sinkConf = idAndConfGenerator.generateConf(); final TestSource source = generateSource(sourceConf); final PhysicalOperator physicalOperator = generateFilterOperator(ocConf, (s) -> true); final PhysicalSink<String> sink = generateSink(sinkConf, result); final ConfigVertex srcVertex = new ConfigVertex(Long.toString(configVertexId.getAndIncrement()), ExecutionVertex.Type.SOURCE, sourceConf); final ConfigVertex ocVertex = new ConfigVertex(Long.toString(configVertexId.getAndIncrement()), ExecutionVertex.Type.OPERATOR, ocConf); final ConfigVertex sinkVertex = new ConfigVertex(Long.toString(configVertexId.getAndIncrement()), ExecutionVertex.Type.SINK, sinkConf); final Tuple<DAG<ConfigVertex, MISTEdge>, ExecutionDag> dagTuple = generateSimpleDag(source, physicalOperator, sink, srcVertex, ocVertex, sinkVertex); final Query query1 = mock(Query.class); final List<String> paths = mock(List.class); queryStarter.start("q1", query1, dagTuple.getKey(), paths); final String data1 = "Hello"; source.send(data1); Assert.assertEquals(Arrays.asList(), result); Assert.assertEquals(1, source.getSourceOutputEmitter().numberOfEvents()); Assert.assertEquals(1, source.getSourceOutputEmitter().processAllEvent()); Assert.assertEquals(Arrays.asList(data1), result); Assert.assertEquals(dagTuple.getKey(), queryIdConfigDagMap.get("q1")); final DAG<ExecutionVertex, MISTEdge> executionDag = dagTuple.getValue().getDag(); Assert.assertTrue(GraphUtils.compareTwoDag(executionDag, srcAndDagMap.get(sourceConf).getDag())); Assert.assertTrue(GraphUtils.compareTwoDag(executionDag, executionVertexDagMap.get(source).getDag())); Assert.assertTrue(GraphUtils.compareTwoDag(executionDag, executionVertexDagMap.get(physicalOperator).getDag())); Assert.assertTrue(GraphUtils.compareTwoDag(executionDag, executionVertexDagMap.get(sink).getDag())); Assert.assertEquals(source, configExecutionVertexMap.get(srcVertex)); Assert.assertEquals(physicalOperator, configExecutionVertexMap.get(ocVertex)); Assert.assertEquals(sink, configExecutionVertexMap.get(sinkVertex)); Assert.assertEquals(1, (int)executionVertexCountMap.get(source)); Assert.assertEquals(1, (int)executionVertexCountMap.get(physicalOperator)); Assert.assertEquals(1, (int)executionVertexCountMap.get(sink)); final Collection<ExecutionDag> expectedDags = new HashSet<>(); expectedDags.add(srcAndDagMap.get(sourceConf)); 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);  }