SamzaSqlQueryParser { public static QueryInfo parseQuery(String sql) { Planner planner = createPlanner(); SqlNode sqlNode; sql = sql.replaceAll(TRAILING_SEMI_COLON_REGEX, ""); try { sqlNode = planner.parse(sql); } catch (SqlParseException e) { String errorMsg = SamzaSqlValidator.formatErrorString(sql, e); LOG.error(errorMsg, e); throw new SamzaException(errorMsg, e); } String sink; String selectQuery; ArrayList<String> sources; if (sqlNode instanceof SqlInsert) { SqlInsert sqlInsert = (SqlInsert) sqlNode; sink = sqlInsert.getTargetTable().toString(); if (sqlInsert.getSource() instanceof SqlSelect) { SqlSelect sqlSelect = (SqlSelect) sqlInsert.getSource(); selectQuery = sqlSelect.toString(); LOG.info("Parsed select query {} from sql {}", selectQuery, sql); sources = getSourcesFromSelectQuery(sqlSelect); } else { String msg = String.format("Sql query is not of the expected format. Select node expected, found %s", sqlInsert.getSource().getClass().toString()); LOG.error(msg); throw new SamzaException(msg); } } else { String msg = String.format("Sql query is not of the expected format. Insert node expected, found %s", sqlNode.getClass().toString()); LOG.error(msg); throw new SamzaException(msg); } return new QueryInfo(selectQuery, sources, sink, sql); } private  SamzaSqlQueryParser(); static QueryInfo parseQuery(String sql);  }
@Test public void testParseQuery() { QueryInfo queryInfo = SamzaSqlQueryParser.parseQuery("insert into log.foo select * from tracking.bar"); Assert.assertEquals("log.foo", queryInfo.getSink()); Assert.assertEquals(1, queryInfo.getSources().size()); Assert.assertEquals("tracking.bar", queryInfo.getSources().get(0)); }
@Test public void testParseGroupyByQuery() { QueryInfo queryInfo = SamzaSqlQueryParser.parseQuery("insert into log.foo select b.pageKey, count(*) from tracking.bar as b group by b.pageKey"); Assert.assertEquals("log.foo", queryInfo.getSink()); Assert.assertEquals(1, queryInfo.getSources().size()); Assert.assertEquals("tracking.bar", queryInfo.getSources().get(0)); }
@Test public void testParseUnNestSubQuery() { QueryInfo queryInfo = SamzaSqlQueryParser.parseQuery("insert into log.foo SELECT * FROM unnest(SELECT int_array_field1 FROM tracking.bar) "); Assert.assertEquals("log.foo", queryInfo.getSink()); Assert.assertEquals(1, queryInfo.getSources().size()); Assert.assertEquals("tracking.bar", queryInfo.getSources().get(0)); }
@Test public void testParseJoinSubQuery() { String sql = "Insert into testavro.enrichedPageViewTopic" + " select p.name as profileName, pv.pageKey" + " from (SELECT * FROM testavro.PAGEVIEW pv1 where pv1.field1='foo') as pv" + " join testavro.PROFILE.`$table` as p" + " on p.id = pv.profileId"; QueryInfo queryInfo = SamzaSqlQueryParser.parseQuery(sql); Assert.assertEquals("testavro.enrichedPageViewTopic", queryInfo.getSink()); Assert.assertEquals(2, queryInfo.getSources().size()); Assert.assertEquals("testavro.PAGEVIEW", queryInfo.getSources().get(0)); Assert.assertEquals("testavro.PROFILE.$table", queryInfo.getSources().get(1)); }
@Test public void testParseJoinUnNestQuery() { String sql = "Insert into testavro.enrichedPageViewTopic" + " select p.name as profileName, pv.pageKey" + " from unnest(SELECT int_array_field1 FROM testavro.PAGEVIEW) as pv" + " join testavro.PROFILE.`$table` as p" + " on p.id = pv.profileId"; QueryInfo queryInfo = SamzaSqlQueryParser.parseQuery(sql); Assert.assertEquals("testavro.enrichedPageViewTopic", queryInfo.getSink()); Assert.assertEquals(2, queryInfo.getSources().size()); Assert.assertEquals("testavro.PAGEVIEW", queryInfo.getSources().get(0)); Assert.assertEquals("testavro.PROFILE.$table", queryInfo.getSources().get(1)); }
@Test public void testParseJoinQuery() { String sql = "Insert into testavro.enrichedPageViewTopic" + " select p.name as profileName, pv.pageKey" + " from testavro.PAGEVIEW as pv" + " join testavro.PROFILE.`$table` as p" + " on p.id = pv.profileId"; QueryInfo queryInfo = SamzaSqlQueryParser.parseQuery(sql); Assert.assertEquals("testavro.enrichedPageViewTopic", queryInfo.getSink()); Assert.assertEquals(2, queryInfo.getSources().size()); Assert.assertEquals("testavro.PAGEVIEW", queryInfo.getSources().get(0)); Assert.assertEquals("testavro.PROFILE.$table", queryInfo.getSources().get(1)); }
@Test public void testParseInvalidQuery() { try { SamzaSqlQueryParser.parseQuery("select * from tracking.bar"); Assert.fail("Expected a samzaException"); } catch (SamzaException e) { } try { SamzaSqlQueryParser.parseQuery("insert into select * from tracking.bar"); Assert.fail("Expected a samzaException"); } catch (SamzaException e) { } try { SamzaSqlQueryParser.parseQuery("insert into log.off select from tracking.bar"); Assert.fail("Expected a samzaException"); } catch (SamzaException e) { } }