@Test public void removeQueryFromTwoExecutionQueriesTest() throws InjectionException, IOException { final List<String> result1 = new LinkedList<>(); 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, result1); 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 String query1Id = "q1"; final String query2Id = "q2"; srcAndDagMap.put(sourceConf1, dagTuple1.getValue()); srcAndDagMap.put(sourceConf2, dagTuple2.getValue()); queryIdConfigDagMap.put(query1Id, dagTuple1.getKey()); queryIdConfigDagMap.put(query2Id, dagTuple2.getKey()); configExecutionVertexMap.put(srcVertex1, src1); configExecutionVertexMap.put(ocVertex1, physicalOp1); configExecutionVertexMap.put(sinkVertex1, sink1); configExecutionVertexMap.put(srcVertex2, src2); configExecutionVertexMap.put(ocVertex2, physicalOp2); configExecutionVertexMap.put(sinkVertex2, sink2); executionDags.add(dagTuple1.getValue()); executionDags.add(dagTuple2.getValue()); executionVertexCountMap.put(src1, 1); executionVertexCountMap.put(physicalOp1, 1); executionVertexCountMap.put(sink1, 1); executionVertexCountMap.put(src2, 1); executionVertexCountMap.put(physicalOp2, 1); executionVertexCountMap.put(sink2, 1); executionVertexDagMap.put(src1, dagTuple1.getValue()); executionVertexDagMap.put(physicalOp1, dagTuple1.getValue()); executionVertexDagMap.put(sink1, dagTuple1.getValue()); executionVertexDagMap.put(src2, dagTuple2.getValue()); executionVertexDagMap.put(physicalOp2, dagTuple2.getValue()); executionVertexDagMap.put(sink2, dagTuple2.getValue()); queryRemover.deleteQuery(query2Id); Assert.assertEquals(1, srcAndDagMap.size()); Assert.assertTrue(GraphUtils.compareTwoDag(dagTuple1.getValue().getDag(), srcAndDagMap.get(src1.getConfiguration()).getDag())); Assert.assertNull(queryIdConfigDagMap.get(query2Id)); Assert.assertEquals(dagTuple1.getKey(), queryIdConfigDagMap.get(query1Id)); Assert.assertEquals(src1, configExecutionVertexMap.get(srcVertex1)); Assert.assertEquals(physicalOp1, configExecutionVertexMap.get(ocVertex1)); Assert.assertEquals(sink1, configExecutionVertexMap.get(sinkVertex1)); Assert.assertNull(configExecutionVertexMap.get(srcVertex2)); Assert.assertNull(configExecutionVertexMap.get(ocVertex2)); Assert.assertNull(configExecutionVertexMap.get(sinkVertex2)); Assert.assertEquals(1, executionDags.values().size()); Assert.assertEquals(1, (int)executionVertexCountMap.get(src1)); Assert.assertEquals(1, (int)executionVertexCountMap.get(physicalOp1)); Assert.assertEquals(1, (int)executionVertexCountMap.get(sink1)); Assert.assertNull(executionVertexCountMap.get(src2)); Assert.assertNull(executionVertexCountMap.get(physicalOp2)); Assert.assertNull(executionVertexCountMap.get(sink2)); Assert.assertTrue(GraphUtils.compareTwoDag(dagTuple1.getValue().getDag(), executionVertexDagMap.get(src1).getDag())); Assert.assertTrue(GraphUtils.compareTwoDag(dagTuple1.getValue().getDag(), executionVertexDagMap.get(physicalOp1).getDag())); Assert.assertTrue(GraphUtils.compareTwoDag(dagTuple1.getValue().getDag(), executionVertexDagMap.get(sink1).getDag())); Assert.assertNull(executionVertexDagMap.get(src2)); Assert.assertNull(executionVertexDagMap.get(physicalOp2)); Assert.assertNull(executionVertexDagMap.get(sink2)); try { src2.send("data"); Assert.fail("Source should be closed"); } catch (final InvalidDataSendException e) { } }
@Override public synchronized void deleteQuery(final String queryId) { synchronized (srcAndDagMap) { final DAG<ConfigVertex, MISTEdge> configDag = queryIdConfigDagMap.remove(queryId); final Collection<ConfigVertex> vertices = configDag.getVertices(); for (final ConfigVertex vertex : vertices) { final ExecutionVertex executionVertex = configExecutionVertexMap.remove(vertex); final int refCount = executionVertexCountMap.get(executionVertex); if (refCount == 1) { final ExecutionDag executionDag = executionVertexDagMap.remove(executionVertex); executionDag.getDag().removeVertex(executionVertex); executionVertexCountMap.remove(executionVertex); if (executionVertex.getType() == ExecutionVertex.Type.SOURCE) { final PhysicalSource src = (PhysicalSource)executionVertex; srcAndDagMap.remove(src.getConfiguration()); try { src.close(); } catch (Exception e) { e.printStackTrace(); } } if (executionDag.getDag().numberOfVertices() == 0) { executionDags.remove(executionDag); } } else { executionVertexCountMap.put(executionVertex, refCount - 1); } } } }
MergeAwareQueryRemover implements QueryRemover { @Override public synchronized void deleteQuery(final String queryId) { synchronized (srcAndDagMap) { final DAG<ConfigVertex, MISTEdge> configDag = queryIdConfigDagMap.remove(queryId); final Collection<ConfigVertex> vertices = configDag.getVertices(); for (final ConfigVertex vertex : vertices) { final ExecutionVertex executionVertex = configExecutionVertexMap.remove(vertex); final int refCount = executionVertexCountMap.get(executionVertex); if (refCount == 1) { final ExecutionDag executionDag = executionVertexDagMap.remove(executionVertex); executionDag.getDag().removeVertex(executionVertex); executionVertexCountMap.remove(executionVertex); if (executionVertex.getType() == ExecutionVertex.Type.SOURCE) { final PhysicalSource src = (PhysicalSource)executionVertex; srcAndDagMap.remove(src.getConfiguration()); try { src.close(); } catch (Exception e) { e.printStackTrace(); } } if (executionDag.getDag().numberOfVertices() == 0) { executionDags.remove(executionDag); } } else { executionVertexCountMap.put(executionVertex, refCount - 1); } } } } }
MergeAwareQueryRemover implements QueryRemover { @Override public synchronized void deleteQuery(final String queryId) { synchronized (srcAndDagMap) { final DAG<ConfigVertex, MISTEdge> configDag = queryIdConfigDagMap.remove(queryId); final Collection<ConfigVertex> vertices = configDag.getVertices(); for (final ConfigVertex vertex : vertices) { final ExecutionVertex executionVertex = configExecutionVertexMap.remove(vertex); final int refCount = executionVertexCountMap.get(executionVertex); if (refCount == 1) { final ExecutionDag executionDag = executionVertexDagMap.remove(executionVertex); executionDag.getDag().removeVertex(executionVertex); executionVertexCountMap.remove(executionVertex); if (executionVertex.getType() == ExecutionVertex.Type.SOURCE) { final PhysicalSource src = (PhysicalSource)executionVertex; srcAndDagMap.remove(src.getConfiguration()); try { src.close(); } catch (Exception e) { e.printStackTrace(); } } if (executionDag.getDag().numberOfVertices() == 0) { executionDags.remove(executionDag); } } else { executionVertexCountMap.put(executionVertex, refCount - 1); } } } } @Inject private  MergeAwareQueryRemover(final QueryIdConfigDagMap queryIdConfigDagMap,
                                 final SrcAndDagMap<Map<String, String>> srcAndDagMap,
                                 final ExecutionDags executionDags,
                                 final ExecutionVertexCountMap executionVertexCountMap,
                                 final ConfigExecutionVertexMap configExecutionVertexMap,
                                 final ExecutionVertexDagMap executionVertexDagMap); }
MergeAwareQueryRemover implements QueryRemover { @Override public synchronized void deleteQuery(final String queryId) { synchronized (srcAndDagMap) { final DAG<ConfigVertex, MISTEdge> configDag = queryIdConfigDagMap.remove(queryId); final Collection<ConfigVertex> vertices = configDag.getVertices(); for (final ConfigVertex vertex : vertices) { final ExecutionVertex executionVertex = configExecutionVertexMap.remove(vertex); final int refCount = executionVertexCountMap.get(executionVertex); if (refCount == 1) { final ExecutionDag executionDag = executionVertexDagMap.remove(executionVertex); executionDag.getDag().removeVertex(executionVertex); executionVertexCountMap.remove(executionVertex); if (executionVertex.getType() == ExecutionVertex.Type.SOURCE) { final PhysicalSource src = (PhysicalSource)executionVertex; srcAndDagMap.remove(src.getConfiguration()); try { src.close(); } catch (Exception e) { e.printStackTrace(); } } if (executionDag.getDag().numberOfVertices() == 0) { executionDags.remove(executionDag); } } else { executionVertexCountMap.put(executionVertex, refCount - 1); } } } } @Inject private  MergeAwareQueryRemover(final QueryIdConfigDagMap queryIdConfigDagMap,
                                 final SrcAndDagMap<Map<String, String>> srcAndDagMap,
                                 final ExecutionDags executionDags,
                                 final ExecutionVertexCountMap executionVertexCountMap,
                                 final ConfigExecutionVertexMap configExecutionVertexMap,
                                 final ExecutionVertexDagMap executionVertexDagMap); @Override synchronized void deleteQuery(final String queryId); @Override void deleteAllQueries(); }
MergeAwareQueryRemover implements QueryRemover { @Override public synchronized void deleteQuery(final String queryId) { synchronized (srcAndDagMap) { final DAG<ConfigVertex, MISTEdge> configDag = queryIdConfigDagMap.remove(queryId); final Collection<ConfigVertex> vertices = configDag.getVertices(); for (final ConfigVertex vertex : vertices) { final ExecutionVertex executionVertex = configExecutionVertexMap.remove(vertex); final int refCount = executionVertexCountMap.get(executionVertex); if (refCount == 1) { final ExecutionDag executionDag = executionVertexDagMap.remove(executionVertex); executionDag.getDag().removeVertex(executionVertex); executionVertexCountMap.remove(executionVertex); if (executionVertex.getType() == ExecutionVertex.Type.SOURCE) { final PhysicalSource src = (PhysicalSource)executionVertex; srcAndDagMap.remove(src.getConfiguration()); try { src.close(); } catch (Exception e) { e.printStackTrace(); } } if (executionDag.getDag().numberOfVertices() == 0) { executionDags.remove(executionDag); } } else { executionVertexCountMap.put(executionVertex, refCount - 1); } } } } @Inject private  MergeAwareQueryRemover(final QueryIdConfigDagMap queryIdConfigDagMap,
                                 final SrcAndDagMap<Map<String, String>> srcAndDagMap,
                                 final ExecutionDags executionDags,
                                 final ExecutionVertexCountMap executionVertexCountMap,
                                 final ConfigExecutionVertexMap configExecutionVertexMap,
                                 final ExecutionVertexDagMap executionVertexDagMap); @Override synchronized void deleteQuery(final String queryId); @Override void deleteAllQueries();  }