QueryExecutor implements Closeable { public StreamJob executeQuery(String query) throws Exception { String queryId = metadataStore.registerQuery(query); JobConfigGenerator jobConfigGenerator = new JobConfigGenerator(queryId, metadataStore); SchemaPlus defaultSchema = getDefaultSchema(); if(!isSamzaSQLSchema(defaultSchema)){ throw new SamzaException("Default schema is not an instance of SamzaSQLSchema"); } addSystems(jobConfigGenerator, getRootSchema()); jobConfigGenerator.setModel(metadataStore.registerCalciteModel(queryId, connection.getModel())); QueryPlanner planner = new QueryPlanner(defaultSchema); SamzaRel queryPlan = planner.getPlan(query); defaultJobProps(jobConfigGenerator, defaultSchema.getName()); jobConfigGenerator.setJobName(queryId); queryPlan.populateJobConfiguration(jobConfigGenerator); return new SamzaSQLJobRunner(jobConfigGenerator.getJobConfig()).run(true); } QueryExecutor(SamzaSQLConnection connection, Config config); StreamJob executeQuery(String query); void close();  }
@Test public void testFilter() throws Exception { String ordersTopic = "Orders"; final String schema = salesSchema(zkServer.connectString(), brokers); System.out.println(schema); MockSamzaSQLConnection samzaSQLConnection = new MockSamzaSQLConnection(schema); final Map<String, String> configurations = new HashMap<>(); configurations.put("samza.sql.metastore.zk.connect", zkServer.connectString()); QueryExecutor queryExecutor = new QueryExecutor(samzaSQLConnection, new MapConfig(configurations)); final StreamJob job = queryExecutor.executeQuery("insert into filtered select * from orders where units > 5"); ApplicationStatus status = job.waitForStatus(ApplicationStatus.Running, 20 * 60000); if (status != null && status == ApplicationStatus.Running) { List<Map<Object, Object>> inputMessages = readMessages("/filter-test.json", "input"); List<KeyedMessage> orderMessages = new ArrayList<KeyedMessage>(); for (Map<Object, Object> msg : inputMessages) { orderMessages.add(createOrderFrom(ordersTopic, msg)); } publish(ordersTopic, orderMessages); waitForTopic("filtered"); verify("filtered", "filtergroup0", new QueryOutputVerifier() { @Override public void verify(KafkaStream<byte[], byte[]> stream) throws Exception { List<Map<Object, Object>> output = readMessages("/filter-test.json", "output"); ConsumerIterator<byte[], byte[]> consumerIterator = stream.iterator(); if(consumerIterator.hasNext()) { GenericRecord filteredOrder = filteredOrderFrom(consumerIterator.next().message()); Assert.assertTrue("Filter query was successful.", filteredOrder.get("orderId").equals(Integer.valueOf((String)output.get(0).get("orderId")))); } } }); } else { Assert.fail("Streaming job is not running after 20 seconds."); } job.kill(); }