From 77412124dd8be7d911fa57bdde14bfdcb4106788 Mon Sep 17 00:00:00 2001 From: Srinivasulu Punuru Date: Wed, 31 Jul 2019 13:31:21 -0700 Subject: [PATCH 1/3] Removing RegEx while parsing the sql statement --- .../apache/samza/sql/util/SamzaSqlQueryParser.java | 13 +++++-------- .../samza/sql/util/TestSamzaSqlQueryParser.java | 1 - .../samza/test/samzasql/TestSamzaSqlEndToEnd.java | 2 +- 3 files changed, 6 insertions(+), 10 deletions(-) diff --git a/samza-sql/src/main/java/org/apache/samza/sql/util/SamzaSqlQueryParser.java b/samza-sql/src/main/java/org/apache/samza/sql/util/SamzaSqlQueryParser.java index e81abfae45..00fdebea5c 100644 --- a/samza-sql/src/main/java/org/apache/samza/sql/util/SamzaSqlQueryParser.java +++ b/samza-sql/src/main/java/org/apache/samza/sql/util/SamzaSqlQueryParser.java @@ -53,12 +53,15 @@ import org.apache.samza.SamzaException; import org.apache.samza.sql.interfaces.SamzaSqlDriver; import org.apache.samza.sql.interfaces.SamzaSqlJavaTypeFactoryImpl; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * Utility class that is used to parse the Samza sql query to figure out the sources, sink etc.. */ public class SamzaSqlQueryParser { + private static final Logger LOG = LoggerFactory.getLogger(SamzaSqlQueryParser.class); private SamzaSqlQueryParser() { } @@ -94,13 +97,6 @@ public String getSql() { } public static QueryInfo parseQuery(String sql) { - - Pattern insertIntoSqlPattern = Pattern.compile("insert into (.*) (select .* from (.*))", Pattern.CASE_INSENSITIVE); - Matcher m = insertIntoSqlPattern.matcher(sql); - if (!m.matches()) { - throw new SamzaException("Invalid query format"); - } - Planner planner = createPlanner(); SqlNode sqlNode; try { @@ -117,7 +113,8 @@ public static QueryInfo parseQuery(String sql) { sink = sqlInsert.getTargetTable().toString(); if (sqlInsert.getSource() instanceof SqlSelect) { SqlSelect sqlSelect = (SqlSelect) sqlInsert.getSource(); - selectQuery = m.group(2); + selectQuery = sqlSelect.toString(); + LOG.info("Parsed select query {} from sql {}", selectQuery, sql); sources = getSourcesFromSelectQuery(sqlSelect); } else { throw new SamzaException("Sql query is not of the expected format"); diff --git a/samza-sql/src/test/java/org/apache/samza/sql/util/TestSamzaSqlQueryParser.java b/samza-sql/src/test/java/org/apache/samza/sql/util/TestSamzaSqlQueryParser.java index f15c2f6914..167b0ded6f 100644 --- a/samza-sql/src/test/java/org/apache/samza/sql/util/TestSamzaSqlQueryParser.java +++ b/samza-sql/src/test/java/org/apache/samza/sql/util/TestSamzaSqlQueryParser.java @@ -31,7 +31,6 @@ public class TestSamzaSqlQueryParser { public void testParseQuery() { QueryInfo queryInfo = SamzaSqlQueryParser.parseQuery("insert into log.foo select * from tracking.bar"); Assert.assertEquals("log.foo", queryInfo.getSink()); - Assert.assertEquals(queryInfo.getSelectQuery(), "select * from tracking.bar", queryInfo.getSelectQuery()); Assert.assertEquals(1, queryInfo.getSources().size()); Assert.assertEquals("tracking.bar", queryInfo.getSources().get(0)); } diff --git a/samza-test/src/test/java/org/apache/samza/test/samzasql/TestSamzaSqlEndToEnd.java b/samza-test/src/test/java/org/apache/samza/test/samzasql/TestSamzaSqlEndToEnd.java index aef79265b3..51f3d93861 100644 --- a/samza-test/src/test/java/org/apache/samza/test/samzasql/TestSamzaSqlEndToEnd.java +++ b/samza-test/src/test/java/org/apache/samza/test/samzasql/TestSamzaSqlEndToEnd.java @@ -266,7 +266,7 @@ public void testEndToEndWithProjection() throws Exception { TestAvroSystemFactory.messages.clear(); Map staticConfigs = SamzaSqlTestConfig.fetchStaticConfigsWithFactories(configs, numMessages); String sql1 = "Insert into testavro.outputTopic(id, long_value) " - + " select id, TIMESTAMPDIFF(HOUR, CURRENT_TIMESTAMP, LOCALTIMESTAMP) + MONTH(CURRENT_DATE) as long_value from testavro.SIMPLE1"; + + " select id, TIMESTAMPDIFF(HOUR, CURRENT_TIMESTAMP(), LOCALTIMESTAMP()) + MONTH(CURRENT_DATE()) as long_value from testavro.SIMPLE1"; List sqlStmts = Arrays.asList(sql1); staticConfigs.put(SamzaSqlApplicationConfig.CFG_SQL_STMTS_JSON, JsonUtil.toJson(sqlStmts)); runApplication(new MapConfig(staticConfigs)); From ccd2cbf56a719705ef8987b50bd43b8d280395b9 Mon Sep 17 00:00:00 2001 From: Srinivasulu Punuru Date: Wed, 31 Jul 2019 13:59:41 -0700 Subject: [PATCH 2/3] Adding few more parser tests --- .../sql/util/TestSamzaSqlQueryParser.java | 48 ++++++++++++++++++- 1 file changed, 47 insertions(+), 1 deletion(-) diff --git a/samza-sql/src/test/java/org/apache/samza/sql/util/TestSamzaSqlQueryParser.java b/samza-sql/src/test/java/org/apache/samza/sql/util/TestSamzaSqlQueryParser.java index 167b0ded6f..a01c32a7b3 100644 --- a/samza-sql/src/test/java/org/apache/samza/sql/util/TestSamzaSqlQueryParser.java +++ b/samza-sql/src/test/java/org/apache/samza/sql/util/TestSamzaSqlQueryParser.java @@ -21,9 +21,9 @@ import org.apache.samza.SamzaException; import org.apache.samza.sql.util.SamzaSqlQueryParser.QueryInfo; +import org.junit.Assert; import org.junit.Test; -import junit.framework.Assert; public class TestSamzaSqlQueryParser { @@ -35,6 +35,52 @@ public void testParseQuery() { 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 = From 1da81704e6692bc48ea7da174c25e8b6715ad95a Mon Sep 17 00:00:00 2001 From: Srinivasulu Punuru Date: Wed, 31 Jul 2019 14:11:02 -0700 Subject: [PATCH 3/3] Adding verbose logging --- .../org/apache/samza/sql/util/SamzaSqlQueryParser.java | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/samza-sql/src/main/java/org/apache/samza/sql/util/SamzaSqlQueryParser.java b/samza-sql/src/main/java/org/apache/samza/sql/util/SamzaSqlQueryParser.java index 00fdebea5c..d2ed991dab 100644 --- a/samza-sql/src/main/java/org/apache/samza/sql/util/SamzaSqlQueryParser.java +++ b/samza-sql/src/main/java/org/apache/samza/sql/util/SamzaSqlQueryParser.java @@ -117,10 +117,16 @@ public static QueryInfo parseQuery(String sql) { LOG.info("Parsed select query {} from sql {}", selectQuery, sql); sources = getSourcesFromSelectQuery(sqlSelect); } else { - throw new SamzaException("Sql query is not of the expected format"); + 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 { - throw new SamzaException("Sql query is not of the expected format"); + 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);