Support EMIT CHANGES for KSQL (#1426)

- Add the EMIT CHANGES syntax used in KSQL.

Author: OlivierCavadenti

src/main/java/net/sf/jsqlparser/statement/select/PlainSelect.java

@@ -53,6 +53,7 @@ public class PlainSelect extends ASTNodeAccessImpl implements SelectBody {
     private String forXmlPath;
     private KSQLWindow ksqlWindow = null;
     private boolean noWait = false;
+    private boolean emitChanges = false;
 
     public boolean isUseBrackets() {
         return useBrackets;
@@ -321,6 +322,14 @@ public void setKsqlWindow(KSQLWindow ksqlWindow) {
         this.ksqlWindow = ksqlWindow;
     }
 
+    public void setEmitChanges(boolean emitChanges) {
+        this.emitChanges = emitChanges;
+    }
+
+    public boolean isEmitChanges() {
+        return emitChanges;
+    }
+
     @Override
     @SuppressWarnings({"PMD.CyclomaticComplexity" , "PMD.ExcessiveMethodLength", "PMD.NPathComplexity"})
     public String toString() {
@@ -400,6 +409,9 @@ public String toString() {
                 sql.append(" HAVING ").append(having);
             }
             sql.append(orderByToString(oracleSiblings, orderByElements));
+            if (emitChanges){
+                sql.append(" EMIT CHANGES");
+            }
             if (limit != null) {
                 sql.append(limit);
             }

src/main/java/net/sf/jsqlparser/util/deparser/SelectDeParser.java

@@ -187,7 +187,9 @@ public void visit(PlainSelect plainSelect) {
             new OrderByDeParser(expressionVisitor, buffer).deParse(plainSelect.isOracleSiblings(),
                     plainSelect.getOrderByElements());
         }
-
+        if (plainSelect.isEmitChanges()){
+            buffer.append(" EMIT CHANGES");
+        }
         if (plainSelect.getLimit() != null) {
             new LimitDeparser(buffer).deParse(plainSelect.getLimit());
         }

src/main/jjtree/net/sf/jsqlparser/parser/JSqlParserCC.jjt

@@ -157,6 +157,7 @@ TOKEN: /* SQL Keywords. prefixed with K_ to avoid name clashes */
 |   <K_TRY_CAST: "TRY_CAST">
 |   <K_CHARACTER:"CHARACTER">
 |   <K_CHANGE:"CHANGE">
+|   <K_CHANGES:"CHANGES">
 |   <K_CHECK:"CHECK">
 |   <K_CHECKPOINT:"CHECKPOINT">
 |   <K_CHAR:"CHAR">
@@ -198,6 +199,7 @@ TOKEN: /* SQL Keywords. prefixed with K_ to avoid name clashes */
 |   <K_DUMP:"DUMP">
 |   <K_DUPLICATE: "DUPLICATE">
 |   <K_ELSE:"ELSE">
+|   <K_EMIT : "EMIT">
 |   <K_ENABLE : "ENABLE">
 |   <K_END:"END">
 |   <K_ESCAPE:"ESCAPE">
@@ -1868,10 +1870,10 @@ PlainSelect PlainSelect() #PlainSelect:
     [ having=Having() { plainSelect.setHaving(having); }]
     [LOOKAHEAD(<K_ORDER> <K_SIBLINGS> <K_BY>) orderByElements = OrderByElements() { plainSelect.setOracleSiblings(true); plainSelect.setOrderByElements(orderByElements);    }   ]
     [LOOKAHEAD(<K_ORDER> <K_BY>) orderByElements = OrderByElements() { plainSelect.setOrderByElements(orderByElements);    }   ]
+    [ <K_EMIT> <K_CHANGES> { plainSelect.setEmitChanges(true); } ]
     [LOOKAHEAD(<K_LIMIT>) limit = LimitWithOffset() { plainSelect.setLimit(limit);    } ]
     [LOOKAHEAD(<K_OFFSET>) offset = Offset() { plainSelect.setOffset(offset);    } ]
 	[LOOKAHEAD(<K_LIMIT>, { limit==null }) limit = LimitWithOffset() { plainSelect.setLimit(limit);    } ]
-
     [LOOKAHEAD(<K_FETCH>) fetch = Fetch() { plainSelect.setFetch(fetch);    } ]
 
     [LOOKAHEAD(2) <K_FOR> <K_UPDATE> { plainSelect.setForUpdate(true); }

src/test/java/net/sf/jsqlparser/statement/select/KSQLTest.java

@@ -142,7 +142,7 @@ public void testKSQLTumblingWindows() throws Exception {
 
         statement = CCJSqlParserUtil.parse(sql);
         System.out.println(statement.toString());
-                Select select = (Select) statement;
+        Select select = (Select) statement;
         PlainSelect plainSelect = (PlainSelect) select.getSelectBody();
         assertTrue(plainSelect.getKsqlWindow().isTumblingWindow());
         assertFalse(plainSelect.getKsqlWindow().isSessionWindow());
@@ -153,4 +153,24 @@ public void testKSQLTumblingWindows() throws Exception {
         assertStatementCanBeDeparsedAs(select, sql, true);
         assertSqlCanBeParsedAndDeparsed(sql, true);
     }
+
+    @Test
+    public void testKSQLEmitChanges() throws Exception {
+        String sql = "SELECT * FROM table1 t1 GROUP BY region.id EMIT CHANGES";
+        Statement statement = CCJSqlParserUtil.parse(sql);
+        Select select = (Select) statement;
+        PlainSelect selectBody = (PlainSelect) select.getSelectBody();
+        assertTrue(selectBody.isEmitChanges());
+        assertSqlCanBeParsedAndDeparsed(sql);
+    }
+
+    @Test
+    public void testKSQLEmitChangesWithLimit() throws Exception {
+        String sql = "SELECT * FROM table1 t1 GROUP BY region.id EMIT CHANGES LIMIT 2";
+        Statement statement = CCJSqlParserUtil.parse(sql);
+        Select select = (Select) statement;
+        PlainSelect selectBody = (PlainSelect) select.getSelectBody();
+        assertTrue(selectBody.isEmitChanges());
+        assertSqlCanBeParsedAndDeparsed(sql);
+    }
 }