Support ksqlDB window grace periods and EMIT FINAL

Author: minleejae

README.md

@@ -141,6 +141,7 @@ and missing syntax gets added on demand — [open an issue](https://github.com/J
 | **DML** | `INSERT` · `UPDATE` · `UPSERT` · `MERGE` · `DELETE` · `TRUNCATE TABLE` |
 | **DDL** | `CREATE …` · `ALTER …` · `DROP …` |
 | **PostgreSQL RLS** | `CREATE POLICY` · `ALTER TABLE … ENABLE`/`DISABLE`/`FORCE`/`NO FORCE ROW LEVEL SECURITY` |
+| **ksqlDB windows** | JOIN `WITHIN`, window `GRACE PERIOD`, and `EMIT CHANGES`/`FINAL` |
 | **Salesforce SOQL** | `INCLUDES` · `EXCLUDES` |
 
 Beyond statement shapes, the grammar handles nested sub-selects, bind parameters (`?`,

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

@@ -9,7 +9,6 @@
  */
 package net.sf.jsqlparser.statement.select;
 
-import java.util.Locale;
 import net.sf.jsqlparser.parser.ASTNodeAccessImpl;
 
 import static net.sf.jsqlparser.statement.select.KSQLWindow.TimeUnit;
@@ -23,9 +22,37 @@ public class KSQLJoinWindow extends ASTNodeAccessImpl {
     private TimeUnit beforeTimeUnit;
     private long afterDuration;
     private TimeUnit afterTimeUnit;
+    private boolean usingBrackets = true;
+    private KSQLWindow.Duration gracePeriod;
+
+    public boolean isUsingBrackets() {
+        return usingBrackets || beforeAfter;
+    }
+
+    public void setUsingBrackets(boolean usingBrackets) {
+        this.usingBrackets = usingBrackets;
+    }
+
+    public KSQLJoinWindow withUsingBrackets(boolean usingBrackets) {
+        setUsingBrackets(usingBrackets);
+        return this;
+    }
+
+    public KSQLWindow.Duration getGracePeriod() {
+        return gracePeriod;
+    }
+
+    public void setGracePeriod(KSQLWindow.Duration gracePeriod) {
+        this.gracePeriod = gracePeriod;
+    }
+
+    public KSQLJoinWindow withGracePeriod(KSQLWindow.Duration gracePeriod) {
+        setGracePeriod(gracePeriod);
+        return this;
+    }
 
     public final static TimeUnit from(String timeUnitStr) {
-        return Enum.valueOf(TimeUnit.class, timeUnitStr.toUpperCase(Locale.ROOT));
+        return TimeUnit.from(timeUnitStr);
     }
 
     public boolean isBeforeAfterWindow() {
@@ -86,11 +113,23 @@ public void setAfterTimeUnit(TimeUnit afterTimeUnit) {
 
     @Override
     public String toString() {
+        StringBuilder builder = new StringBuilder();
+        if (isUsingBrackets()) {
+            builder.append('(');
+        }
         if (isBeforeAfterWindow()) {
-            return "(" + beforeDuration + " " + beforeTimeUnit + ", " + afterDuration + " "
-                    + afterTimeUnit + ")";
+            builder.append(beforeDuration).append(' ').append(beforeTimeUnit)
+                    .append(", ").append(afterDuration).append(' ').append(afterTimeUnit);
+        } else {
+            builder.append(duration).append(' ').append(timeUnit);
+        }
+        if (isUsingBrackets()) {
+            builder.append(')');
+        }
+        if (gracePeriod != null) {
+            builder.append(" GRACE PERIOD ").append(gracePeriod);
         }
-        return "(" + duration + " " + timeUnit + ")";
+        return builder.toString();
     }
 
     public KSQLJoinWindow withDuration(long duration) {

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

@@ -21,6 +21,47 @@ public class KSQLWindow extends ASTNodeAccessImpl {
     private TimeUnit sizeTimeUnit;
     private long advanceDuration;
     private TimeUnit advanceTimeUnit;
+    private Duration gracePeriod;
+
+    /** A non-negative duration with its SQL time unit, including an explicit zero. */
+    public static final class Duration implements java.io.Serializable {
+        private final long value;
+        private final TimeUnit timeUnit;
+
+        public Duration(long value, TimeUnit timeUnit) {
+            if (value < 0) {
+                throw new IllegalArgumentException("Duration must not be negative");
+            }
+            this.value = value;
+            this.timeUnit = java.util.Objects.requireNonNull(timeUnit, "timeUnit");
+        }
+
+        public long getValue() {
+            return value;
+        }
+
+        public TimeUnit getTimeUnit() {
+            return timeUnit;
+        }
+
+        @Override
+        public String toString() {
+            return value + " " + timeUnit;
+        }
+    }
+
+    public Duration getGracePeriod() {
+        return gracePeriod;
+    }
+
+    public void setGracePeriod(Duration gracePeriod) {
+        this.gracePeriod = gracePeriod;
+    }
+
+    public KSQLWindow withGracePeriod(Duration gracePeriod) {
+        setGracePeriod(gracePeriod);
+        return this;
+    }
 
     public KSQLWindow() {}
 
@@ -82,14 +123,20 @@ public void setAdvanceTimeUnit(TimeUnit advanceTimeUnit) {
 
     @Override
     public String toString() {
+        StringBuilder builder = new StringBuilder();
         if (isHoppingWindow()) {
-            return "HOPPING (" + "SIZE " + sizeDuration + " " + sizeTimeUnit + ", " +
-                    "ADVANCE BY " + advanceDuration + " " + advanceTimeUnit + ")";
+            builder.append("HOPPING (SIZE ").append(sizeDuration).append(' ').append(sizeTimeUnit)
+                    .append(", ADVANCE BY ").append(advanceDuration).append(' ')
+                    .append(advanceTimeUnit);
         } else if (isSessionWindow()) {
-            return "SESSION (" + sizeDuration + " " + sizeTimeUnit + ")";
+            builder.append("SESSION (").append(sizeDuration).append(' ').append(sizeTimeUnit);
         } else {
-            return "TUMBLING (" + "SIZE " + sizeDuration + " " + sizeTimeUnit + ")";
+            builder.append("TUMBLING (SIZE ").append(sizeDuration).append(' ').append(sizeTimeUnit);
+        }
+        if (gracePeriod != null) {
+            builder.append(", GRACE PERIOD ").append(gracePeriod);
         }
+        return builder.append(')').toString();
     }
 
     public KSQLWindow withSizeDuration(long sizeDuration) {

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

@@ -55,7 +55,8 @@ public class PlainSelect extends Select {
     private MySqlSqlCacheFlags mySqlCacheFlag = null;
     private String forXmlPath;
     private KSQLWindow ksqlWindow = null;
-    private boolean emitChanges = false;
+    private EmitMode emitMode = EmitMode.NONE;
+
     private List<WindowDefinition> windowDefinitions;
     /**
      * @see <a href=
@@ -68,6 +69,10 @@ public class PlainSelect extends Select {
     private Table intoTempTable = null;
     private List<UpdateSet> settings = null;
 
+    public enum EmitMode {
+        NONE, CHANGES, FINAL
+    }
+
     public PlainSelect() {}
 
     public PlainSelect(FromItem fromItem) {
@@ -500,12 +505,32 @@ public void setKsqlWindow(KSQLWindow ksqlWindow) {
         this.ksqlWindow = ksqlWindow;
     }
 
+    public EmitMode getEmitMode() {
+        return emitMode;
+    }
+
+    public void setEmitMode(EmitMode emitMode) {
+        this.emitMode = java.util.Objects.requireNonNull(emitMode, "emitMode");
+    }
+
+    public PlainSelect withEmitMode(EmitMode emitMode) {
+        setEmitMode(emitMode);
+        return this;
+    }
+
+    public StringBuilder appendEmitClauseTo(StringBuilder builder) {
+        if (emitMode != EmitMode.NONE) {
+            builder.append(" EMIT ").append(emitMode);
+        }
+        return builder;
+    }
+
     public boolean isEmitChanges() {
-        return emitChanges;
+        return emitMode == EmitMode.CHANGES;
     }
 
     public void setEmitChanges(boolean emitChanges) {
-        this.emitChanges = emitChanges;
+        emitMode = emitChanges ? EmitMode.CHANGES : EmitMode.NONE;
     }
 
     public List<WindowDefinition> getWindowDefinitions() {
@@ -634,9 +659,7 @@ public StringBuilder appendSelectBodyTo(StringBuilder builder) {
             builder.append(windowDefinitions.stream().map(WindowDefinition::toString)
                     .collect(joining(", ")));
         }
-        if (emitChanges) {
-            builder.append(" EMIT CHANGES");
-        }
+        appendEmitClauseTo(builder);
         if (intoTempTable != null) {
             builder.append(" INTO TEMP ").append(intoTempTable);
         }

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

@@ -354,9 +354,7 @@ public <S> StringBuilder visit(PlainSelect plainSelect, S context) {
             builder.append(plainSelect.getOption());
         }
 
-        if (plainSelect.isEmitChanges()) {
-            builder.append(" EMIT CHANGES");
-        }
+        plainSelect.appendEmitClauseTo(builder);
         if (plainSelect.getLimitBy() != null) {
             new LimitDeparser(expressionVisitor, builder).deParse(plainSelect.getLimitBy());
         }