feat: enable new emit-final implementation #9141

mjsax · 2022-05-21T01:04:05Z

No description provided.

CLAassistant · 2022-05-21T01:04:10Z

All committers have signed the CLA.

mjsax · 2022-05-21T01:05:34Z

ksqldb-execution/src/main/java/io/confluent/ksql/execution/windows/HoppingWindowExpression.java

@@ -80,7 +92,7 @@ public String toString() {
        + "ADVANCE BY " + advanceBy
        + retention.map(w -> " , RETENTION " + w).orElse("")
        + gracePeriod.map(g -> " , GRACE PERIOD " + g).orElse("")
-        + " ) ";
+        + " ) " + emitStrategy.map(Enum::toString).orElse("");


This is a nasty hack -- did not find a better way to preserve the information.

For other readers: this "hack" is being added so that deserialization works properly here: #9141 (comment)

vcrfxia

Thanks @mjsax -- some initial comments/questions inline.

vcrfxia · 2022-05-23T23:27:16Z

ksqldb-common/src/main/java/io/confluent/ksql/serde/WindowInfo.java

@@ -73,12 +85,14 @@ public boolean equals(final Object o) {
      return false;
    }
    final WindowInfo that = (WindowInfo) o;
+    // we omit `emitStrategy`


Why do we omit this from the comparator?

WindowInfo should actually only describe the window itself and some code relies on it -- emit-strategy is not really a window property and it's just bolted on her as it's hard to get a hold on it otherwise.

Because emit-strategy could be set on one instance but not on another, while both instances describe the same window, we omit it here to not break the comparison.

(IIRC)

ksqldb-engine/src/main/java/io/confluent/ksql/ddl/commands/CreateSourceFactory.java

vcrfxia · 2022-05-23T23:30:04Z

ksqldb-parser/src/main/java/io/confluent/ksql/parser/json/WindowExpressionDeserializer.java

+
+    final KsqlWindowExpression windowExpression;
+
+    if (extendedWindowExpression.endsWith("FINAL")) {


Instead of this hack, can we update the serializer/deserializer to serialize an additional, optional field to specify the emitStrategy? Ideally there'd be a way to deserialize a window object that is either represented as a raw string (legacy) or as an object (new).

mjsax · 2022-05-21T01:05:42Z

ksqldb-execution/src/main/java/io/confluent/ksql/execution/windows/SessionWindowExpression.java

@@ -62,7 +73,7 @@ public String toString() {
    return " SESSION ( " + gap
        + retention.map(w -> " , RETENTION " + w).orElse("")
        + gracePeriod.map(g -> " , GRACE PERIOD " + g).orElse("")
-        + " ) ";
+        + " ) " + emitStrategy.map(Enum::toString).orElse("");


This is a nasty hack -- did not find a better way to preserve the information.

mjsax · 2022-05-21T01:05:49Z

...db-execution/src/main/java/io/confluent/ksql/execution/windows/TumblingWindowExpression.java

@@ -65,7 +73,7 @@ public String toString() {
    return " TUMBLING ( SIZE " + size
        + retention.map(w -> " , RETENTION " + w).orElse("")
        + gracePeriod.map(g -> " , GRACE PERIOD " + g).orElse("")
-        + " ) ";
+        + " ) " + emitStrategy.map(Enum::toString).orElse("");


This is a nasty hack -- did not find a better way to preserve the information.

mjsax · 2022-05-21T01:06:03Z

ksqldb-parser/src/main/java/io/confluent/ksql/parser/json/WindowExpressionDeserializer.java

+
+    final KsqlWindowExpression windowExpression;
+
+    if (extendedWindowExpression.endsWith("FINAL")) {


This is a nasty hack -- did not find a better way to preserve the information.

mjsax · 2022-06-17T16:07:21Z

ksqldb-common/src/main/java/io/confluent/ksql/config/KsqlConfigResolver.java

+  private static boolean isInternalStreamsConfig(final String key) {
+    // add more internal configs on demand
+    return key.equals(InternalConfig.EMIT_INTERVAL_MS_KSTREAMS_OUTER_JOIN_SPURIOUS_RESULTS_FIX)
+        || key.equals(InternalConfig.EMIT_INTERVAL_MS_KSTREAMS_WINDOWED_AGGREGATION);


Need this one to run QTTs...

Is this only meant for QTT or are we expecting users to want/be able to set this for their own queries too? (Guessing the latter but want to check.)

Also, the first config (outer join spurious results fix emit interval) is unrelated to this PR, right? Is that another config we think users might want to set?

Both are internal KS configs -- they control how often we try to emit results -- the other configs is for left/outer stream-stream joins (yes, it's unrelated to this PR, but serves the exact same purpose for a different operator).

We need them for QTT to ensure we alway try to emit result (both configs work on wall-clock time); otherwise, in QTT results might get "stuck" and are not emitted. I remember that for stream-stream join back in the days, we did not add test for all scenarios because we did not have access to this config.

Because both are internal, we don't really expect users to set them (with the exception that there is a perf problem and we tell them to change it). But we should be able for set the configs IMHO -- otherwise, the guard they provide for KS does not work for ksqlDB.

There are a few more internal configs, and we could consider to add them here, too. Also happy to remove the second config from this PR and we do some follow up.

That makes sense. Just to confirm: users will be able to set these internal streams configs on their ksql queries (and have the configs take effect) after this change, right?

I think it's fine to leave the second config in -- we're essentially saying it's safe for users to set these configs on their ksql queries if they want to.

Technically, yes.

Of course, both configs are "internal" meaning it's not documented anywhere (on purpose) and user most likely don't know about them (and should not need to know about them in general).

mjsax · 2022-06-17T16:30:14Z

ksqldb-engine/src/main/java/io/confluent/ksql/planner/LogicalPlanner.java

+          throw new KsqlException("EMIT FINAL is only supported for windowed aggregations.");
+        }
+        if (refinementInfo.getOutputRefinement() != OutputRefinement.FINAL_PERSISTENT) {
+          currentNode = buildSuppressNode(


Only add old suppress() for old/historic plans. New FINAL_PERSISTENT does not use suppress()

vcrfxia

Thanks @mjsax -- LGTM at a high level. Some questions inline.

vcrfxia · 2022-06-23T22:11:51Z

ksqldb-common/src/main/java/io/confluent/ksql/config/KsqlConfigResolver.java

+  private static boolean isInternalStreamsConfig(final String key) {
+    // add more internal configs on demand
+    return key.equals(InternalConfig.EMIT_INTERVAL_MS_KSTREAMS_OUTER_JOIN_SPURIOUS_RESULTS_FIX)
+        || key.equals(InternalConfig.EMIT_INTERVAL_MS_KSTREAMS_WINDOWED_AGGREGATION);


Is this only meant for QTT or are we expecting users to want/be able to set this for their own queries too? (Guessing the latter but want to check.)

Also, the first config (outer join spurious results fix emit interval) is unrelated to this PR, right? Is that another config we think users might want to set?

vcrfxia · 2022-06-23T22:14:43Z

ksqldb-common/src/main/java/io/confluent/ksql/parser/OutputRefinement.java

   *
   * <p>For a table, this would mean output only the finalized result per-key.
   *
   * <p>For a stream, all events are final, so all are output.
   */
-  FINAL;
+  FINAL,


Is this old value still needed after this PR? It should be obsolete, right? Existing queries have suppress nodes in their plans, and new queries will only ever set FINAL_PERSISTENT instead.

🤔 -- fair point. Could be that we don't need it...

Wondering if we should remove it? Or if we should re-purpose it?

The distinction would only be internal, right? As in, users will never see FINAL vs FINAL_PERSISTENT, and we also don't write anything of the sort into the command topic. If so, I think either's fine since we'd always be able to change it in the future. My vote is to repurpose FINAL if you feel confident that it's safe, else replace FINAL with FINAL_PERSISTENT since we know for sure that's safe.

Yes, internal only, there is just EMIT FINAL syntax for the user and if this is merged it should always use FINAL_PERSISTENT internally.

My vote is to repurpose

Ack. Will look into it. Was not 100% sure if it would be save, but seems it should be. Better to cleanup right away.

vcrfxia · 2022-06-23T22:16:06Z

ksqldb-common/src/main/java/io/confluent/ksql/serde/WindowInfo.java

@@ -73,12 +85,14 @@ public boolean equals(final Object o) {
      return false;
    }
    final WindowInfo that = (WindowInfo) o;
+    // we omit `emitStrategy`


This is just a "philosophical" choice (saying that conceptually a window info doesn't care about emit strategy), rather than a practical choice, right? As in, nothing breaks if we decide to include emitStrategy in the equals/hashCode methods, does it?

Some stuff did break... I originally included it and only remove it later and added the comment (cannot remember from the top of my head what did break though).

Interesting! I'd be curious to know what broke, but if it's difficult to find out again then don't worry about it.

Could not reproduce it -- maybe something change in the mean time. Added it to both equals and hashCode.

vcrfxia · 2022-06-23T22:16:22Z

ksqldb-common/src/test/java/io/confluent/ksql/serde/WindowInfoTest.java

+        .addEqualityGroup(
+            WindowInfo.of(TUMBLING, Optional.of(Duration.ofMillis(1020)), Optional.of(OutputRefinement.CHANGES)),
+            WindowInfo.of(TUMBLING, Optional.of(Duration.ofMillis(1020)), Optional.of(OutputRefinement.FINAL)),
+            WindowInfo.of(TUMBLING, Optional.of(Duration.ofMillis(1020)), Optional.of(OutputRefinement.FINAL_PERSISTENT))


nit: empty optional should compare equal in this group too

vcrfxia · 2022-06-23T22:18:07Z

ksqldb-engine/src/main/java/io/confluent/ksql/analyzer/RewrittenAnalysis.java

+    // we only need to rewrite if we have a window expression and if we use emit final
+    if (!windowExpression.isPresent()
+        || !refinementInfo.isPresent()
+        || !refinementInfo.get().isFinal()) { // can we test for isFinal? or memory vs rocks?


What's the concern here? Are you saying that users of in-memory stores wouldn't want to enable the new emit strategy? (Probably shouldn't affect ksql since ksql always uses rocks, just trying to understand.)

We test for !isFinal() that returns !true for FINAL (in-memory) || FINAL_PERSISTENT (rocks)?

The question is, if we need to test for only !FINAL_PERSISTENT instead?

Given your other comment about FINAL maybe being obsolete, the answer is "no". However, if we should return the unmodified window expression only for FINAL (but not for FINAL_PERSISTENT) the answer might be "yes".

Got it, I initially misunderstood what you meant by "memory vs rocks". I would hope that only new queries go through the RewrittenAnalysis so the two should be equivalent. It's been a while since I've worked with this part of the code, though.

vcrfxia · 2022-06-23T22:19:42Z

ksqldb-functional-tests/src/test/resources/query-validation-tests/suppress.json

@@ -3,7 +3,8 @@
    {
      "name": "should emit final result immediately at window end if grace is specified as zero",
      "properties": {
-        "ksql.suppress.enabled": true
+        "ksql.suppress.enabled": true,


We need new historic plans for these existing QTTs with this change, right? The query plans should've changed to no longer have the suppress.

Correct. I did not yet create historical plan on purpose, as it "pollutes" the PR (and I need to wait for the session window KS PR anyway)

ksqldb-engine/src/main/java/io/confluent/ksql/ddl/commands/CommandFactories.java

mjsax · 2022-06-29T18:18:41Z

ksqldb-common/src/main/java/io/confluent/ksql/serde/WindowInfo.java

@@ -73,20 +85,35 @@ public boolean equals(final Object o) {
      return false;
    }
    final WindowInfo that = (WindowInfo) o;
+
+    final Optional<OutputRefinement> thisEmit = emitStrategy.isPresent()


@vcrfxia -- I hope this make sense? The idea is to make empty and changes equal (also cf hashCode below)

Makes sense but I have a code style question -- if empty is always the same as EMIT CHANGES, then can be change the constructor to store OutputRefinement rather than Optional<OutputRefinement> and use emitStrategy.orElse(OutputRefinement.CHANGES) in the constructor, in order to simplify these methods here?

I did consider it, but was not sure -- as you bring it up, too, seems to be the better choice. Let me update it.

vcrfxia · 2022-06-29T23:13:38Z

ksqldb-common/src/main/java/io/confluent/ksql/serde/WindowInfo.java

@@ -73,20 +85,35 @@ public boolean equals(final Object o) {
      return false;
    }
    final WindowInfo that = (WindowInfo) o;
+
+    final Optional<OutputRefinement> thisEmit = emitStrategy.isPresent()


Makes sense but I have a code style question -- if empty is always the same as EMIT CHANGES, then can be change the constructor to store OutputRefinement rather than Optional<OutputRefinement> and use emitStrategy.orElse(OutputRefinement.CHANGES) in the constructor, in order to simplify these methods here?

vcrfxia · 2022-06-29T23:15:44Z

ksqldb-engine/src/test/java/io/confluent/ksql/planner/LogicalPlannerTest.java

@@ -193,6 +192,7 @@ public void testSimpleRightJoinFilterLogicalPlan() {
    assertThat(rightSource.getSources().get(0), instanceOf(PreJoinRepartitionNode.class));
  }

+  @Ignore


Looks like the latest commit adds this Ignore annotation to a number of unit tests. I'm guessing you have plans to address these?

vcrfxia

Woohoo, thanks @mjsax ! LGTM with some non-blocking questions/comments inline.

vcrfxia · 2022-07-07T19:53:55Z

ksqldb-common/src/main/java/io/confluent/ksql/serde/WindowInfo.java

@@ -64,6 +74,10 @@ public Optional<Duration> getSize() {
    return size;
  }

+  public OutputRefinement getEmitStrategy() {


Does the serialize/deserialize to/from JSON loop work for this, now that this getter is updated to return OutputRefinement while the constructor still accepts Optional<OutputRefinement>? (Hopefully yes and I assume yes if QTTs are passing, just curious since I haven't encountered the situation before.)

🤔 -- Interesting question... To me it just did not make sense to return an Optional if we know that it's always set... Did not think about JSON (de)serialization. But I agree, it seem to work so we should be good?

vcrfxia · 2022-07-07T20:00:35Z

ksqldb-functional-tests/src/test/resources/query-validation-tests/suppress.json

@@ -20,9 +20,6 @@
    },
    {
      "name": "should not emit final result before window end if grace is specified as zero",
-      "properties": {
-        "ksql.suppress.enabled": true


How come most of the tests have "ksql.streams.__emit.interval.ms.kstreams.windowed.aggregation__": 0 added, but this one and the one immediately below it don't?

Guess it still passed because we expect no output (and thus I missed it), but I think we should set it to zero to not accidentally mask a bug.

Great catch!

vcrfxia · 2022-07-07T20:01:08Z

ksqldb-functional-tests/src/test/resources/sql-tests/query-upgrades/filters.sql

@@ -328,10 +328,9 @@ CREATE OR REPLACE TABLE b AS SELECT id, COUNT(*)
 ----------------------------------------------------------------------------------------------------
 --@test: add filter to Suppress
 [email protected]: io.confluent.ksql.util.KsqlException
-[email protected]: Upgrades not yet supported for TableSuppress
+[email protected]: Upgrades not yet supported for StreamWindowedAggregate


I'm fine simply deleting this test rather than updating the expected error message. There's already another one which expects Upgrades not yet supported for StreamWindowedAggregate.

Maybe @agavra can chime in if this test still adds value or not?

yeah we can probably remove this test - the idea was that if you had a suppress it would fail regardless of what was being aggregated, but I don't think it adds much value since suppress only makes sense with windowed aggregates anyway (which we already test for above)

Follow up PR to remove this test: #9308

mjsax requested a review from a team as a code owner May 21, 2022 01:04

mjsax commented May 21, 2022

View reviewed changes

vcrfxia reviewed May 23, 2022

View reviewed changes

mjsax commented May 24, 2022

View reviewed changes

mjsax commented Jun 17, 2022

View reviewed changes

vcrfxia reviewed Jun 23, 2022

View reviewed changes

mjsax commented Jun 29, 2022

View reviewed changes

vcrfxia reviewed Jun 29, 2022

View reviewed changes

mjsax added 8 commits July 5, 2022 09:10

feat: [WIP] enable new emit-final implemenation

58aa967

WIP

7838aec

minor

d3132a4

Github comments

5978aea

fix equals/hashcode

6c1e50d

fixes

dcd7a85

enable sessions windows

7a8ba81

generate schemas

c69a0fd

mjsax changed the title ~~feat: [WIP] enable new emit-final implementation~~ feat: enable new emit-final implementation Jul 5, 2022

mjsax added 2 commits July 6, 2022 11:15

Add historic plans

913cc7a

fix test

7ec592e

vcrfxia approved these changes Jul 7, 2022

View reviewed changes

github comments

22c60f5

mjsax merged commit 2af201f into confluentinc:master Jul 7, 2022

mjsax deleted the emit-final branch July 7, 2022 21:15

mjsax mentioned this pull request Jul 18, 2022

test: remove stale tests #9308

Merged


		final KsqlWindowExpression windowExpression;

		if (extendedWindowExpression.endsWith("FINAL")) {

feat: enable new emit-final implementation #9141

feat: enable new emit-final implementation #9141

Conversation

mjsax commented May 21, 2022

CLAassistant commented May 21, 2022 • edited by cla-assistant bot Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vcrfxia left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mjsax Jun 24, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vcrfxia left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mjsax Jun 28, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mjsax Jun 24, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vcrfxia left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

CLAassistant commented May 21, 2022 •

edited by cla-assistant bot

Loading

mjsax Jun 24, 2022 •

edited

Loading

mjsax Jun 28, 2022 •

edited

Loading

mjsax Jun 24, 2022 •

edited

Loading