From e4226a0177fd273949ea75c83cb68c2fc8b985e5 Mon Sep 17 00:00:00 2001 From: Minjae Lee Date: Thu, 10 Sep 2026 21:13:39 +0900 Subject: [PATCH 1/2] Support ksqlDB window grace periods and EMIT FINAL --- README.md | 1 + .../statement/select/KSQLJoinWindow.java | 49 +++++- .../statement/select/KSQLWindow.java | 55 ++++++- .../statement/select/PlainSelect.java | 35 +++- .../util/deparser/SelectDeParser.java | 4 +- .../net/sf/jsqlparser/parser/JSqlParserCC.jjt | 133 +++++++++------ .../select/KSQLWindowOptionsTest.java | 153 ++++++++++++++++++ 7 files changed, 360 insertions(+), 70 deletions(-) create mode 100644 src/test/java/net/sf/jsqlparser/statement/select/KSQLWindowOptionsTest.java diff --git a/README.md b/README.md index 8ad991b09..f59569628 100644 --- a/README.md +++ b/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 (`?`, diff --git a/src/main/java/net/sf/jsqlparser/statement/select/KSQLJoinWindow.java b/src/main/java/net/sf/jsqlparser/statement/select/KSQLJoinWindow.java index 5afbecd3d..16daac002 100644 --- a/src/main/java/net/sf/jsqlparser/statement/select/KSQLJoinWindow.java +++ b/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) { diff --git a/src/main/java/net/sf/jsqlparser/statement/select/KSQLWindow.java b/src/main/java/net/sf/jsqlparser/statement/select/KSQLWindow.java index ea50d4cb6..eae98921f 100644 --- a/src/main/java/net/sf/jsqlparser/statement/select/KSQLWindow.java +++ b/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) { diff --git a/src/main/java/net/sf/jsqlparser/statement/select/PlainSelect.java b/src/main/java/net/sf/jsqlparser/statement/select/PlainSelect.java index 1f289effe..f84bf6ea9 100644 --- a/src/main/java/net/sf/jsqlparser/statement/select/PlainSelect.java +++ b/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 windowDefinitions; /** * @see 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 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); } diff --git a/src/main/java/net/sf/jsqlparser/util/deparser/SelectDeParser.java b/src/main/java/net/sf/jsqlparser/util/deparser/SelectDeParser.java index b506b0b27..f78e3d71d 100644 --- a/src/main/java/net/sf/jsqlparser/util/deparser/SelectDeParser.java +++ b/src/main/java/net/sf/jsqlparser/util/deparser/SelectDeParser.java @@ -354,9 +354,7 @@ public 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()); } diff --git a/src/main/jjtree/net/sf/jsqlparser/parser/JSqlParserCC.jjt b/src/main/jjtree/net/sf/jsqlparser/parser/JSqlParserCC.jjt index 860ab3821..f32cf997a 100644 --- a/src/main/jjtree/net/sf/jsqlparser/parser/JSqlParserCC.jjt +++ b/src/main/jjtree/net/sf/jsqlparser/parser/JSqlParserCC.jjt @@ -1167,6 +1167,12 @@ public class CCJSqlParser extends AbstractJSqlParser { // String-literal alias: SELECT col 'myAlias' if (kind == S_CHAR_LITERAL) return true; + // WITHIN followed by a duration starts a ksqlDB JOIN window. + if (kind == K_WITHIN && (getToken(2).kind == S_LONG + || getToken(2).kind == OPENING_BRACKET && getToken(3).kind == S_LONG)) { + return false; + } + // OPTION (...) introduces a query hint clause, not an alias if (kind == K_OPTION && getToken(2).kind == OPENING_BRACKET) { return false; @@ -6168,7 +6174,9 @@ PlainSelect PlainSelect() #PlainSelect: [ LOOKAHEAD(2) interpolateElements = InterpolateClause() { plainSelect.setInterpolate(interpolateElements); } ] [ LOOKAHEAD(2) forClause = ForClause() {plainSelect.setForClause(forClause);} ] [ LOOKAHEAD(2) optionClause = OptionClause() { plainSelect.setOption(optionClause); } ] - [ LOOKAHEAD(2) { plainSelect.setEmitChanges(true); } ] + [ LOOKAHEAD(2) + ( { plainSelect.setEmitMode(PlainSelect.EmitMode.CHANGES); } + | { plainSelect.setEmitMode(PlainSelect.EmitMode.FINAL); } ) ] // Parse the LIMIT row count once (this accepts a parenthesized subquery too), then // optionally attach ClickHouse's `LIMIT ... BY ...`. Checking BY right after the limit // expression avoids a numeric LOOKAHEAD, which cannot see past a long parenthesized @@ -7180,7 +7188,7 @@ Join JoinerExpression() #JoinerExpression: [ LOOKAHEAD(2) ( - [ "(" joinWindow = JoinWindow() ")" {join.setJoinWindow(joinWindow);} ] + [ joinWindow = KSQLJoinWindowClause() {join.setJoinWindow(joinWindow);} ] ( onExpression=Expression() { join.addOnExpression(onExpression); } ( LOOKAHEAD({ getToken(1).kind == K_ON && !isInsertOnConflictAhead() }) onExpression=Expression() { join.addOnExpression(onExpression); } )* ) @@ -7200,74 +7208,95 @@ Join JoinerExpression() #JoinerExpression: } +/** Shared duration parsing for JOIN, aggregation windows, and grace periods. */ +KSQLWindow.Duration KSQLDuration(): +{ + Token value; + Token unit; +} +{ + value= ( unit= | unit= ) + { return new KSQLWindow.Duration(Long.parseLong(value.image), KSQLWindow.TimeUnit.from(unit.image)); } +} + +KSQLWindow.Duration KSQLGracePeriod(): +{ + KSQLWindow.Duration duration; +} +{ + AccessKeyword("GRACE") AccessKeyword("PERIOD") duration=KSQLDuration() + { return duration; } +} + +KSQLJoinWindow KSQLJoinWindowClause(): +{ + KSQLJoinWindow window; + KSQLWindow.Duration duration; +} +{ + ( + "(" window=JoinWindow() ")" + | + duration=KSQLDuration() { + window = new KSQLJoinWindow().withDuration(duration.getValue()) + .withTimeUnit(duration.getTimeUnit()).withUsingBrackets(false); + } + ) + [ LOOKAHEAD({ isKeywordAhead("GRACE") }) duration=KSQLGracePeriod() { window.setGracePeriod(duration); } ] + { return window; } +} + KSQLJoinWindow JoinWindow(): { - KSQLJoinWindow retval = new KSQLJoinWindow(); - boolean beforeAfter; - Token beforeDurationToken = null; - Token beforeTimeUnitToken = null; - Token afterDurationToken = null; - Token afterTimeUnitToken = null; + KSQLJoinWindow window = new KSQLJoinWindow(); + KSQLWindow.Duration before; + KSQLWindow.Duration after = null; } { - beforeDurationToken= (beforeTimeUnitToken= | beforeTimeUnitToken=) - [ "," afterDurationToken= (afterTimeUnitToken= | afterTimeUnitToken=) ] + before=KSQLDuration() + [ "," after=KSQLDuration() ] { - if (afterDurationToken == null) { - retval.setDuration(Long.parseLong(beforeDurationToken.image)); - retval.setTimeUnit(KSQLWindow.TimeUnit.from(beforeTimeUnitToken.image)); - retval.setBeforeAfterWindow(false); - return retval; + if (after == null) { + window.setDuration(before.getValue()); + window.setTimeUnit(before.getTimeUnit()); + } else { + window.setBeforeDuration(before.getValue()); + window.setBeforeTimeUnit(before.getTimeUnit()); + window.setAfterDuration(after.getValue()); + window.setAfterTimeUnit(after.getTimeUnit()); + window.setBeforeAfterWindow(true); } - retval.setBeforeDuration(Long.parseLong(beforeDurationToken.image)); - retval.setBeforeTimeUnit(KSQLWindow.TimeUnit.from(beforeTimeUnitToken.image)); - retval.setAfterDuration(Long.parseLong(afterDurationToken.image)); - retval.setAfterTimeUnit(KSQLWindow.TimeUnit.from(afterTimeUnitToken.image)); - retval.setBeforeAfterWindow(true); - return retval; + return window; } } KSQLWindow KSQLWindowClause(): { - KSQLWindow retval = null; - Token sizeDurationToken = null; - Token sizeTimeUnitToken = null; - Token advanceDurationToken = null; - Token advanceTimeUnitToken = null; + KSQLWindow window = new KSQLWindow(); + KSQLWindow.Duration size; + KSQLWindow.Duration advance; + KSQLWindow.Duration grace; } { - { - retval=new KSQLWindow(); - retval.setHoppingWindow(false); - retval.setSessionWindow(false); - retval.setTumblingWindow(false); - } ( - "(" - sizeDurationToken= sizeTimeUnitToken= "," - advanceDurationToken= advanceTimeUnitToken= ")" - { - retval.setHoppingWindow(true); - } | - "(" sizeDurationToken= sizeTimeUnitToken= ")" - { - retval.setSessionWindow(true); - } | - "(" sizeDurationToken= sizeTimeUnitToken= ")" - { - retval.setTumblingWindow(true); + "(" size=KSQLDuration() + "," advance=KSQLDuration() { + window.setHoppingWindow(true); + window.setAdvanceDuration(advance.getValue()); + window.setAdvanceTimeUnit(advance.getTimeUnit()); } + | + "(" size=KSQLDuration() { window.setSessionWindow(true); } + | + "(" size=KSQLDuration() { window.setTumblingWindow(true); } ) + [ "," grace=KSQLGracePeriod() { window.setGracePeriod(grace); } ] + ")" { - retval.setSizeDuration(Long.parseLong(sizeDurationToken.image)); - retval.setSizeTimeUnit(KSQLWindow.TimeUnit.from(sizeTimeUnitToken.image)); - if (advanceDurationToken != null) { - retval.setAdvanceDuration(Long.parseLong(advanceDurationToken.image)); - retval.setAdvanceTimeUnit(KSQLWindow.TimeUnit.from(advanceTimeUnitToken.image)); - } - return retval; + window.setSizeDuration(size.getValue()); + window.setSizeTimeUnit(size.getTimeUnit()); + return window; } } diff --git a/src/test/java/net/sf/jsqlparser/statement/select/KSQLWindowOptionsTest.java b/src/test/java/net/sf/jsqlparser/statement/select/KSQLWindowOptionsTest.java new file mode 100644 index 000000000..e4c682f44 --- /dev/null +++ b/src/test/java/net/sf/jsqlparser/statement/select/KSQLWindowOptionsTest.java @@ -0,0 +1,153 @@ +/*- + * #%L + * JSQLParser library + * %% + * Copyright (C) 2004 - 2019 JSQLParser + * %% + * Dual licensed under GNU LGPL 2.1 or Apache License 2.0 + * #L% + */ +package net.sf.jsqlparser.statement.select; + +import static net.sf.jsqlparser.test.TestUtils.assertSqlCanBeParsedAndDeparsed; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import java.util.List; +import net.sf.jsqlparser.JSQLParserException; +import net.sf.jsqlparser.parser.CCJSqlParserUtil; +import net.sf.jsqlparser.statement.Statements; +import net.sf.jsqlparser.statement.select.KSQLWindow.Duration; +import net.sf.jsqlparser.statement.select.KSQLWindow.TimeUnit; +import net.sf.jsqlparser.statement.select.PlainSelect.EmitMode; +import net.sf.jsqlparser.util.deparser.StatementDeParser; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; +import org.junit.jupiter.params.provider.ValueSource; + +class KSQLWindowOptionsTest { + private PlainSelect roundTrip(String sql) throws Exception { + PlainSelect select = (PlainSelect) assertSqlCanBeParsedAndDeparsed(sql); + StringBuilder builder = new StringBuilder(); + select.accept(new StatementDeParser(builder), null); + assertEquals(select.toString(), builder.toString()); + assertEquals(select.toString(), CCJSqlParserUtil.parse(builder.toString()).toString()); + return select; + } + + @ParameterizedTest + @EnumSource(TimeUnit.class) + void joinsAcceptAllTimeUnitsAndPreserveBrackets(TimeUnit unit) throws Exception { + for (boolean brackets : List.of(false, true)) { + String within = brackets ? "(1 " + unit + ")" : "1 " + unit; + PlainSelect select = roundTrip("SELECT a.id FROM a INNER JOIN b WITHIN " + within + + " GRACE PERIOD 0 SECONDS ON a.id = b.id EMIT CHANGES"); + KSQLJoinWindow window = select.getJoins().get(0).getJoinWindow(); + assertEquals(1, window.getDuration()); + assertEquals(unit, window.getTimeUnit()); + assertEquals(brackets, window.isUsingBrackets()); + assertEquals(0, window.getGracePeriod().getValue()); + assertEquals(TimeUnit.SECONDS, window.getGracePeriod().getTimeUnit()); + assertTrue(select.isEmitChanges()); + } + } + + @ParameterizedTest + @EnumSource(TimeUnit.class) + void aggregationWindowsShareTimeUnitsAndGrace(TimeUnit unit) throws Exception { + for (String shape : List.of("TUMBLING (SIZE 20 ", "SESSION (20 ", + "HOPPING (SIZE 20 ")) { + String spec = shape + unit + + (shape.startsWith("HOPPING") ? ", ADVANCE BY 5 " + unit : "") + + ", GRACE PERIOD 2 " + unit + ")"; + PlainSelect select = + roundTrip("SELECT item_id, SUM(quantity) FROM orders WINDOW " + spec + + " GROUP BY item_id EMIT FINAL LIMIT 10"); + KSQLWindow window = select.getKsqlWindow(); + assertEquals(20, window.getSizeDuration()); + assertEquals(unit, window.getSizeTimeUnit()); + assertEquals(2, window.getGracePeriod().getValue()); + assertEquals(unit, window.getGracePeriod().getTimeUnit()); + assertEquals(EmitMode.FINAL, select.getEmitMode()); + assertFalse(select.isEmitChanges()); + if (shape.startsWith("HOPPING")) { + assertEquals(5, window.getAdvanceDuration()); + assertEquals(unit, window.getAdvanceTimeUnit()); + } + } + } + + @Test + void preservesAsymmetricWindowsAndFollowingJoins() throws Exception { + PlainSelect select = roundTrip( + "SELECT a.id FROM a JOIN b WITHIN (1 HOUR, 5 MINUTES) GRACE PERIOD 2 SECONDS ON a.id = b.id JOIN c WITHIN 10 SECONDS ON c.id = a.id EMIT CHANGES"); + assertEquals(2, select.getJoins().size()); + KSQLJoinWindow window = select.getJoins().get(0).getJoinWindow(); + assertTrue(window.isBeforeAfterWindow()); + assertEquals(1, window.getBeforeDuration()); + assertEquals(TimeUnit.HOUR, window.getBeforeTimeUnit()); + assertEquals(5, window.getAfterDuration()); + assertEquals(TimeUnit.MINUTES, window.getAfterTimeUnit()); + assertNull(select.getJoins().get(1).getJoinWindow().getGracePeriod()); + } + + @Test + void parsesOriginalReproducers() throws Exception { + roundTrip( + "select count(*) from log_data_testtopic_1b138324 t1 inner join TestTopic t2 WITHIN 1 MINUTES on t1.email = t2.email EMIT CHANGES"); + PlainSelect select = roundTrip( + "SELECT count(*), sum('id'), avg('id') FROM TestTopic WINDOW TUMBLING (SIZE 60 SECONDS) EMIT FINAL"); + assertEquals(EmitMode.FINAL, select.getEmitMode()); + } + + @Test + void retainsLegacyBuilderSemanticsAndAllowsMutation() throws Exception { + PlainSelect select = (PlainSelect) CCJSqlParserUtil.parse("SELECT * FROM t"); + assertEquals(EmitMode.NONE, select.getEmitMode()); + select.setEmitChanges(true); + assertEquals(EmitMode.CHANGES, select.getEmitMode()); + assertTrue(select.toString().endsWith("EMIT CHANGES")); + select.setEmitMode(EmitMode.FINAL); + assertTrue(select.toString().endsWith("EMIT FINAL")); + select.setEmitChanges(false); + assertEquals("SELECT * FROM t", select.toString()); + KSQLJoinWindow window = new KSQLJoinWindow().withDuration(5).withTimeUnit(TimeUnit.MINUTES); + assertEquals("(5 MINUTES)", window.toString()); + window.withUsingBrackets(false).withGracePeriod(new Duration(0, TimeUnit.SECONDS)); + assertEquals("5 MINUTES GRACE PERIOD 0 SECONDS", window.toString()); + window.setGracePeriod(null); + assertEquals("5 MINUTES", window.toString()); + KSQLWindow aggregate = + new KSQLWindow().withSizeDuration(10).withSizeTimeUnit(TimeUnit.SECONDS) + .withGracePeriod(new Duration(2, TimeUnit.MINUTES)); + assertEquals("TUMBLING (SIZE 10 SECONDS, GRACE PERIOD 2 MINUTES)", aggregate.toString()); + aggregate.setGracePeriod(null); + assertEquals("TUMBLING (SIZE 10 SECONDS)", aggregate.toString()); + assertThrows(IllegalArgumentException.class, () -> new Duration(-1, TimeUnit.SECONDS)); + } + + @Test + void keepsFollowingStatementsAndKeywordIdentifiers() throws Exception { + Statements statements = CCJSqlParserUtil.parseStatements( + "SELECT item_id, COUNT(*) FROM orders WINDOW TUMBLING (SIZE 1 HOUR) GROUP BY item_id EMIT FINAL; SELECT grace, period FROM t;"); + assertEquals(2, statements.size()); + assertEquals("SELECT grace, period FROM t", statements.get(1).toString()); + roundTrip("SELECT * FROM a JOIN b within ON a.id = within.id"); + roundTrip("SELECT * FROM a JOIN b AS within ON a.id = within.id"); + } + + @ParameterizedTest + @ValueSource(strings = { + "SELECT * FROM a JOIN b WITHIN 1 BANANAS ON a.id = b.id", + "SELECT * FROM a JOIN b WITHIN 1 HOURS GRACE 2 MINUTES ON a.id = b.id", + "SELECT * FROM a JOIN b WITHIN 1 HOURS, 2 HOURS ON a.id = b.id", + "SELECT * FROM t WINDOW TUMBLING (SIZE 1 HOUR, GRACE PERIOD -1 SECONDS)", + "SELECT * FROM t WINDOW TUMBLING (SIZE 1 HOUR, GRACE PERIOD 1 SECOND, GRACE PERIOD 2 SECONDS)", + "SELECT * FROM t EMIT FINAL CHANGES"}) + void rejectsMalformedWindowOptions(String sql) { + assertThrows(JSQLParserException.class, () -> CCJSqlParserUtil.parse(sql)); + } +} From e701ded41346a3a86412a9466db42e33a4f9bdd6 Mon Sep 17 00:00:00 2001 From: Minjae Lee Date: Thu, 10 Sep 2026 21:25:12 +0900 Subject: [PATCH 2/2] Group ksqlDB window support with queries in the support table --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index f59569628..9cf291dbb 100644 --- a/README.md +++ b/README.md @@ -138,10 +138,10 @@ and missing syntax gets added on demand — [open an issue](https://github.com/J | | Statements | |---|---| | **Queries** | `SELECT` · `WITH …` · Piped SQL | +| **ksqlDB windows** | JOIN `WITHIN`, window `GRACE PERIOD`, and `EMIT CHANGES`/`FINAL` | | **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 (`?`,