PolicyInterpreter { public static PolicyExecutionPlan parseExecutionPlan(String policyDefinition, Map<String, StreamDefinition> inputStreamDefinitions) throws Exception { Preconditions.checkNotNull(inputStreamDefinitions, "No inputStreams to connect from"); return parseExecutionPlan(SiddhiDefinitionAdapter.buildSiddhiExecutionPlan(policyDefinition, inputStreamDefinitions)); }  static PolicyParseResult parse(String executionPlanQuery); static PolicyExecutionPlan parseExecutionPlan(String policyDefinition, Map<String, StreamDefinition> inputStreamDefinitions); static PolicyExecutionPlan parseExecutionPlan(String executionPlanQuery); static PolicyValidationResult validate(PolicyDefinition policy, Map<String, StreamDefinition> allDefinitions);  }
@Test public void testParseSingleStreamPolicyQuery() throws Exception { PolicyExecutionPlan executionPlan = PolicyInterpreter.parseExecutionPlan("from HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX#window.externalTime(timestamp, 2 min) " + "select cmd, user, count() as total_count group by cmd,user insert into HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX_OUT"); Assert.assertEquals("HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX", executionPlan.getInputStreams().keySet().toArray()[0]); Assert.assertEquals("HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX_OUT", executionPlan.getOutputStreams().keySet().toArray()[0]); Assert.assertEquals(1, executionPlan.getStreamPartitions().size()); Assert.assertEquals(2*60*1000,executionPlan.getStreamPartitions().get(0).getSortSpec().getWindowPeriodMillis()); }
@Test public void testParseSingleStreamPolicyWithPattern() throws Exception { PolicyExecutionPlan executionPlan = PolicyInterpreter.parseExecutionPlan( "from e1=Stream1[price >= 20] -> e2=Stream2[price >= e1.price] \n" + "select e1.symbol as symbol, e2.price as price, e1.price+e2.price as total_price \n" + "group by symbol, company insert into OutStream"); Assert.assertTrue(executionPlan.getInputStreams().containsKey("Stream1")); Assert.assertTrue(executionPlan.getInputStreams().containsKey("Stream2")); Assert.assertTrue(executionPlan.getOutputStreams().containsKey("OutStream")); Assert.assertEquals(StreamPartition.Type.GROUPBY,executionPlan.getStreamPartitions().get(0).getType()); Assert.assertArrayEquals(new String[]{"symbol","company"},executionPlan.getStreamPartitions().get(0).getColumns().toArray()); Assert.assertEquals(StreamPartition.Type.GROUPBY,executionPlan.getStreamPartitions().get(1).getType()); Assert.assertArrayEquals(new String[]{"symbol","company"},executionPlan.getStreamPartitions().get(1).getColumns().toArray()); }
@Test public void testParseSingleStreamPolicyQueryWithMultiplePartitionUsingLargerWindow() throws Exception { PolicyExecutionPlan executionPlan = PolicyInterpreter.parseExecutionPlan( "from HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX#window.externalTime(timestamp, 1 min) " + "select cmd,user, count() as total_count group by cmd,user insert into HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX_OUT_1;" + "from HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX#window.externalTime(timestamp, 1 hour) " + "select cmd,user, count() as total_count group by cmd,user insert into HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX_OUT_2;" ); Assert.assertEquals("HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX", executionPlan.getInputStreams().keySet().toArray()[0]); Assert.assertTrue(executionPlan.getOutputStreams().containsKey("HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX_OUT_1")); Assert.assertTrue(executionPlan.getOutputStreams().containsKey("HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX_OUT_2")); Assert.assertEquals(1, executionPlan.getStreamPartitions().size()); Assert.assertEquals(60*60*1000, executionPlan.getStreamPartitions().get(0).getSortSpec().getWindowPeriodMillis()); }
@Test(expected = SiddhiAppValidationException.class) public void testParseSingleStreamPolicyQueryWithConflictPartition() throws Exception { PolicyExecutionPlan executionPlan = PolicyInterpreter.parseExecutionPlan( "from HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX#window.externalTime(timestamp, 5 min) " + "select cmd, count() as total_count group by cmd,user insert into HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX_OUT_1;" + "from HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX#window.externalTime(timestamp, 2 min) " + "select user, count() as total_count group by cmd,user insert into HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX_OUT_2;" ); Assert.assertEquals("HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX", executionPlan.getInputStreams().keySet().toArray()[0]); Assert.assertTrue(executionPlan.getOutputStreams().containsKey("HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX_OUT_1")); Assert.assertTrue(executionPlan.getOutputStreams().containsKey("HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX_OUT_2")); Assert.assertEquals(2, executionPlan.getStreamPartitions().size()); Assert.assertEquals(5*60*1000, executionPlan.getStreamPartitions().get(0).getSortSpec().getWindowPeriodMillis()); }
@Test public void testParseTwoStreamPolicyQueryWithMultiplePartition() throws Exception { PolicyExecutionPlan executionPlan = PolicyInterpreter.parseExecutionPlan( "from HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX_1#window.externalTime(timestamp, 1 min) " + "select cmd,user, count() as total_count group by cmd,user insert into HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX_OUT_1;" + "from HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX_2#window.externalTime(timestamp, 1 hour) " + "select cmd,user, count() as total_count group by cmd,user insert into HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX_OUT_2;" ); Assert.assertTrue(executionPlan.getInputStreams().containsKey("HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX_1")); Assert.assertTrue(executionPlan.getInputStreams().containsKey("HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX_2")); Assert.assertTrue(executionPlan.getOutputStreams().containsKey("HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX_OUT_1")); Assert.assertTrue(executionPlan.getOutputStreams().containsKey("HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX_OUT_2")); Assert.assertEquals(2, executionPlan.getStreamPartitions().size()); Assert.assertEquals(60*1000, executionPlan.getStreamPartitions().get(0).getSortSpec().getWindowPeriodMillis()); Assert.assertEquals(60*60*1000, executionPlan.getStreamPartitions().get(1).getSortSpec().getWindowPeriodMillis()); }
@Test public void testParseTwoStreamPolicyQueryWithSinglePartition() throws Exception { PolicyExecutionPlan executionPlan = PolicyInterpreter.parseExecutionPlan( "from HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX_1#window.externalTime(timestamp, 1 min) " + "select cmd,user, count() as total_count group by cmd,user insert into HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX_OUT_1;" + "from HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX_2 select * insert into HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX_OUT_2;" ); Assert.assertTrue(executionPlan.getInputStreams().containsKey("HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX_1")); Assert.assertTrue(executionPlan.getInputStreams().containsKey("HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX_2")); Assert.assertTrue(executionPlan.getOutputStreams().containsKey("HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX_OUT_1")); Assert.assertTrue(executionPlan.getOutputStreams().containsKey("HDFS_AUDIT_LOG_ENRICHED_STREAM_SANDBOX_OUT_2")); Assert.assertEquals(2, executionPlan.getStreamPartitions().size()); Assert.assertEquals(60*1000, executionPlan.getStreamPartitions().get(0).getSortSpec().getWindowPeriodMillis()); Assert.assertEquals(StreamPartition.Type.SHUFFLE, executionPlan.getStreamPartitions().get(1).getType()); }
@Test public void testParseTwoStreamPolicyQueryInnerJoin() throws Exception { PolicyExecutionPlan executionPlan = PolicyInterpreter.parseExecutionPlan( "from TickEvent[symbol=='EBAY']#window.length(2000) " + "join NewsEvent#window.externalTime(timestamp, 1000 sec) \n" + "select * insert into JoinStream" ); Assert.assertTrue(executionPlan.getInputStreams().containsKey("TickEvent")); Assert.assertTrue(executionPlan.getInputStreams().containsKey("NewsEvent")); Assert.assertTrue(executionPlan.getOutputStreams().containsKey("JoinStream")); Assert.assertEquals(StreamPartition.Type.SHUFFLE, executionPlan.getStreamPartitions().get(0).getType()); Assert.assertNotNull(executionPlan.getStreamPartitions().get(0).getSortSpec()); Assert.assertEquals(1000*1000, executionPlan.getStreamPartitions().get(0).getSortSpec().getWindowPeriodMillis()); Assert.assertEquals(StreamPartition.Type.SHUFFLE, executionPlan.getStreamPartitions().get(1).getType()); Assert.assertNull(executionPlan.getStreamPartitions().get(1).getSortSpec()); }
@Test public void testParseTwoStreamPolicyQueryInnerJoinWithCondition() throws Exception { PolicyExecutionPlan executionPlan = PolicyInterpreter.parseExecutionPlan( "from TickEvent[symbol=='EBAY']#window.length(2000) as t unidirectional \n" + "join NewsEvent#window.externalTime(timestamp, 1000 sec) as n \n" + "on TickEvent.symbol == NewsEvent.company \n" + "insert into JoinStream " ); Assert.assertTrue(executionPlan.getInputStreams().containsKey("TickEvent")); Assert.assertTrue(executionPlan.getInputStreams().containsKey("NewsEvent")); Assert.assertTrue(executionPlan.getOutputStreams().containsKey("JoinStream")); Assert.assertEquals(StreamPartition.Type.SHUFFLE, executionPlan.getStreamPartitions().get(0).getType()); Assert.assertNotNull(executionPlan.getStreamPartitions().get(0).getSortSpec()); Assert.assertEquals(1000*1000, executionPlan.getStreamPartitions().get(0).getSortSpec().getWindowPeriodMillis()); Assert.assertEquals(StreamPartition.Type.GROUPBY, executionPlan.getStreamPartitions().get(1).getType()); Assert.assertNull(executionPlan.getStreamPartitions().get(1).getSortSpec()); }
@Test public void testParseTwoStreamPolicyQueryInnerJoinWithConditionHavingAlias() throws Exception { PolicyExecutionPlan executionPlan = PolicyInterpreter.parseExecutionPlan( "from TickEvent[symbol=='EBAY']#window.length(2000) as t unidirectional \n" + "join NewsEvent#window.externalTime(timestamp, 1000 sec) as n \n" + "on t.symbol == n.company \n" + "insert into JoinStream " ); Assert.assertTrue(executionPlan.getInputStreams().containsKey("TickEvent")); Assert.assertTrue(executionPlan.getInputStreams().containsKey("NewsEvent")); Assert.assertTrue(executionPlan.getOutputStreams().containsKey("JoinStream")); Assert.assertEquals(StreamPartition.Type.SHUFFLE, executionPlan.getStreamPartitions().get(0).getType()); Assert.assertNotNull(executionPlan.getStreamPartitions().get(0).getSortSpec()); Assert.assertEquals(1000*1000, executionPlan.getStreamPartitions().get(0).getSortSpec().getWindowPeriodMillis()); Assert.assertEquals(StreamPartition.Type.GROUPBY, executionPlan.getStreamPartitions().get(1).getType()); Assert.assertNull(executionPlan.getStreamPartitions().get(1).getSortSpec()); }
@Test(expected = DefinitionNotExistException.class) public void testParseTwoStreamPolicyQueryInnerJoinWithConditionHavingNotFoundAlias() throws Exception { PolicyInterpreter.parseExecutionPlan( "from TickEvent[symbol=='EBAY']#window.length(2000) as t unidirectional \n" + "join NewsEvent#window.externalTime(timestamp, 1000 sec) as n \n" + "on t.symbol == NOT_EXIST_ALIAS.company \n" + "insert into JoinStream " ); }