SamzaSqlValidator { public void validate(List<String> sqlStmts) throws SamzaSqlValidatorException { for (String sql: sqlStmts) { SamzaSqlApplicationConfig sqlConfig = SamzaSqlDslConverter.getSqlConfig(Collections.singletonList(sql), config); QueryPlanner planner = SamzaSqlDslConverter.getQueryPlanner(sqlConfig); SamzaSqlQueryParser.QueryInfo qinfo = SamzaSqlQueryParser.parseQuery(sql); RelRoot relRoot; try { relRoot = planner.plan(qinfo.getSelectQuery()); } catch (SamzaException e) { throw new SamzaSqlValidatorException(String.format("Validation failed for sql stmt:\n%s\n with the following" + " error: \n%s\n", sql, e), e); } String sink = qinfo.getSink(); validate(relRoot, sink, sqlConfig.getRelSchemaProviders().get(sink), sqlConfig.getSamzaRelConverters().get(sink)); } } SamzaSqlValidator(Config config); void validate(List<String> sqlStmts); static String formatErrorString(String query, Exception e);  }
@Test public void testBasicValidation() throws SamzaSqlValidatorException { Map<String, String> config = SamzaSqlTestConfig.fetchStaticConfigsWithFactories(1); config.put(SamzaSqlApplicationConfig.CFG_SQL_STMT, "Insert into testavro.outputTopic select id, true as bool_value, name as string_value" + " from testavro.level1.level2.SIMPLE1 as s where s.id = 1"); Config samzaConfig = SamzaSqlApplicationRunner.computeSamzaConfigs(true, new MapConfig(config)); List<String> sqlStmts = fetchSqlFromConfig(config); new SamzaSqlValidator(samzaConfig).validate(sqlStmts); }
@Test public void testRepeatedTwiceFieldsValidation() throws SamzaSqlValidatorException { Map<String, String> config = SamzaSqlTestConfig.fetchStaticConfigsWithFactories(1); config.put(SamzaSqlApplicationConfig.CFG_SQL_STMT, "Insert into testavro.outputTopic select false as bool_value, c.* from testavro.COMPLEX1 as c"); Config samzaConfig = SamzaSqlApplicationRunner.computeSamzaConfigs(true, new MapConfig(config)); List<String> sqlStmts = fetchSqlFromConfig(config); new SamzaSqlValidator(samzaConfig).validate(sqlStmts); }
@Test (expected = SamzaSqlValidatorException.class) public void testRepeatedThriceFieldsValidation() throws SamzaSqlValidatorException { Map<String, String> config = SamzaSqlTestConfig.fetchStaticConfigsWithFactories(1); config.put(SamzaSqlApplicationConfig.CFG_SQL_STMT, "Insert into testavro.outputTopic select id, bool_value, true as bool_value, c.* from testavro.COMPLEX1 as c"); Config samzaConfig = SamzaSqlApplicationRunner.computeSamzaConfigs(true, new MapConfig(config)); List<String> sqlStmts = fetchSqlFromConfig(config); new SamzaSqlValidator(samzaConfig).validate(sqlStmts); }
@Test (expected = SamzaSqlValidatorException.class) public void testIllegitFieldEndingInZeroValidation() throws SamzaSqlValidatorException { Map<String, String> config = SamzaSqlTestConfig.fetchStaticConfigsWithFactories(1); config.put(SamzaSqlApplicationConfig.CFG_SQL_STMT, "Insert into testavro.outputTopic select id, true as bool_value, false as non_existing_name0" + " from testavro.level1.level2.SIMPLE1 as s where s.id = 1"); Config samzaConfig = SamzaSqlApplicationRunner.computeSamzaConfigs(true, new MapConfig(config)); List<String> sqlStmts = fetchSqlFromConfig(config); new SamzaSqlValidator(samzaConfig).validate(sqlStmts); }
@Test public void testLegitFieldEndingInZeroValidation() throws SamzaSqlValidatorException { Map<String, String> config = SamzaSqlTestConfig.fetchStaticConfigsWithFactories(1); config.put(SamzaSqlApplicationConfig.CFG_SQL_STMT, "Insert into testavro.outputTopic" + " select id, bool_value, float_value0 from testavro.COMPLEX1"); Config samzaConfig = SamzaSqlApplicationRunner.computeSamzaConfigs(true, new MapConfig(config)); List<String> sqlStmts = fetchSqlFromConfig(config); new SamzaSqlValidator(samzaConfig).validate(sqlStmts); }
@Test (expected = SamzaSqlValidatorException.class) public void testNonExistingOutputField() throws SamzaSqlValidatorException { Map<String, String> config = SamzaSqlTestConfig.fetchStaticConfigsWithFactories(1); config.put(SamzaSqlApplicationConfig.CFG_SQL_STMT, "Insert into testavro.outputTopic(id) select id, name as strings_value" + " from testavro.level1.level2.SIMPLE1 as s where s.id = 1"); Config samzaConfig = SamzaSqlApplicationRunner.computeSamzaConfigs(true, new MapConfig(config)); List<String> sqlStmts = fetchSqlFromConfig(config); new SamzaSqlValidator(samzaConfig).validate(sqlStmts); }
@Test (expected = SamzaException.class) public void testNonExistingUdf() throws SamzaSqlValidatorException { Map<String, String> config = SamzaSqlTestConfig.fetchStaticConfigsWithFactories(1); config.put(SamzaSqlApplicationConfig.CFG_SQL_STMT, "Insert into testavro.outputTopic(id) select NonExistingUdf(name) as string_value" + " from testavro.level1.level2.SIMPLE1 as s where s.id = 1"); Config samzaConfig = SamzaSqlApplicationRunner.computeSamzaConfigs(true, new MapConfig(config)); List<String> sqlStmts = fetchSqlFromConfig(config); new SamzaSqlValidator(samzaConfig).validate(sqlStmts); }
@Test (expected = SamzaSqlValidatorException.class) public void testSelectAndOutputValidationFailure() throws SamzaSqlValidatorException { Map<String, String> config = SamzaSqlTestConfig.fetchStaticConfigsWithFactories(1); config.put(SamzaSqlApplicationConfig.CFG_SQL_STMT, "Insert into testavro.outputTopic(id) select name as long_value" + " from testavro.level1.level2.SIMPLE1 as s where s.id = 1"); Config samzaConfig = SamzaSqlApplicationRunner.computeSamzaConfigs(true, new MapConfig(config)); List<String> sqlStmts = fetchSqlFromConfig(config); new SamzaSqlValidator(samzaConfig).validate(sqlStmts); }
@Test (expected = SamzaException.class) public void testValidationStreamTableLeftJoinWithWhere() throws SamzaSqlValidatorException { Map<String, String> config = SamzaSqlTestConfig.fetchStaticConfigsWithFactories(configs, 1); String sql = "Insert into testavro.enrichedPageViewTopic(profileName, pageKey) select p.name as profileName, pv.pageKey" + " from testavro.PAGEVIEW as pv left join testavro.PROFILE.`$table` as p where p.id = pv.profileId"; config.put(SamzaSqlApplicationConfig.CFG_SQL_STMT, sql); Config samzaConfig = SamzaSqlApplicationRunner.computeSamzaConfigs(true, new MapConfig(config)); List<String> sqlStmts = fetchSqlFromConfig(config); new SamzaSqlValidator(samzaConfig).validate(sqlStmts); }
@Test (expected = SamzaException.class) public void testUnsupportedOperator() throws SamzaSqlValidatorException { Map<String, String> config = SamzaSqlTestConfig.fetchStaticConfigsWithFactories(configs, 1); String sql = "Insert into testavro.pageViewCountTopic(jobName, pageKey, `count`)" + " select 'SampleJob' as jobName, pv.pageKey, count(*) as `count`" + " from testavro.PAGEVIEW as pv" + " where pv.pageKey = 'job' or pv.pageKey = 'inbox'" + " group bys (pv.pageKey)"; config.put(SamzaSqlApplicationConfig.CFG_SQL_STMT, sql); Config samzaConfig = SamzaSqlApplicationRunner.computeSamzaConfigs(true, new MapConfig(config)); List<String> sqlStmts = fetchSqlFromConfig(config); new SamzaSqlValidator(samzaConfig).validate(sqlStmts); }
@Test public void testNonDefaultButNullableField() throws SamzaSqlValidatorException { Map<String, String> config = SamzaSqlTestConfig.fetchStaticConfigsWithFactories(1); config.put(SamzaSqlApplicationConfig.CFG_SQL_STMT, "Insert into testavro.outputTopic(id) select id, NOT(id = 5) as bool_value from testavro.SIMPLE1"); Config samzaConfig = SamzaSqlApplicationRunner.computeSamzaConfigs(true, new MapConfig(config)); List<String> sqlStmts = fetchSqlFromConfig(config); new SamzaSqlValidator(samzaConfig).validate(sqlStmts); }
@Test public void testDefaultWithNonNullableField() { Map<String, String> config = SamzaSqlTestConfig.fetchStaticConfigsWithFactories(configs, 1); config.put(SamzaSqlApplicationConfig.CFG_SQL_STMT, "Insert into testavro.outputTopic(id) select id from testavro.SIMPLE1"); Config samzaConfig = SamzaSqlApplicationRunner.computeSamzaConfigs(true, new MapConfig(config)); List<String> sqlStmts = fetchSqlFromConfig(config); try { new SamzaSqlValidator(samzaConfig).validate(sqlStmts); } catch (SamzaSqlValidatorException e) { Assert.assertTrue(e.getMessage().contains("Non-optional field 'bool_value' in output schema is missing")); return; } Assert.fail("Validation test has failed."); }
@Test public void testNonDefaultOutputField() { Map<String, String> config = SamzaSqlTestConfig.fetchStaticConfigsWithFactories(configs, 1); String sql = "Insert into testavro.outputTopic " + " select NOT(id = 5) as bool_value, CASE WHEN id IN (5, 6, 7) THEN CAST('foo' AS VARCHAR) WHEN id < 5 THEN CAST('bars' AS VARCHAR) ELSE NULL END as string_value from testavro.SIMPLE1"; config.put(SamzaSqlApplicationConfig.CFG_SQL_STMT, sql); Config samzaConfig = SamzaSqlApplicationRunner.computeSamzaConfigs(true, new MapConfig(config)); List<String> sqlStmts = fetchSqlFromConfig(config); try { new SamzaSqlValidator(samzaConfig).validate(sqlStmts); } catch (SamzaSqlValidatorException e) { Assert.assertTrue(e.getMessage().contains("Non-optional field 'id' in output schema is missing")); return; } Assert.fail("Validation test has failed."); }