KAFKA-13494: WindowKeyQuery and WindowRangeQuery #11567

patrickstuedi · 2021-12-06T14:01:10Z

No description provided.

vvcephei

Hey @patrickstuedi , it looks like your PR is still in progress, so I just gave it a quick pass. I hope this helps!

streams/src/main/java/org/apache/kafka/streams/query/WindowKeyQuery.java

streams/src/main/java/org/apache/kafka/streams/query/WindowRangeQuery.java

vvcephei · 2021-12-21T22:17:17Z

streams/src/main/java/org/apache/kafka/streams/state/internals/InMemorySessionStore.java

+        if (query instanceof WindowRangeQuery) {
+            @SuppressWarnings("unchecked") final WindowRangeQuery<Bytes, byte[]> windowRangeQuery = (WindowRangeQuery<Bytes, byte[]>) query;
+            if (windowRangeQuery.getKey().isPresent()) {
+                final Bytes key = windowRangeQuery.getKey().get();
+                final KeyValueIterator<Windowed<Bytes>, byte[]> keyValueIterator = this.fetch(key);
+                @SuppressWarnings("unchecked") final R result = (R) keyValueIterator;
+                final QueryResult<R> queryResult = QueryResult.forResult(result);
+                return queryResult;
+            }
+        }
+
+        return null;
+
+        /*
        return StoreQueryUtils.handleBasicQueries(
-            query,
-            positionBound,
-            collectExecutionInfo,
-            this,
-            position,
-            context.taskId().partition()
+                query,
+                positionBound,
+                collectExecutionInfo,
+                this,
+                position,
+                context.taskId().partition()
        );
+
+         */


I assume that the commented code is just a temporary state while you're finalizing the PR.

Since executing these queries doesn't require any specialization to the concrete store type, just the WindowStore and SessionStore interfaces, I think you might want to move the logic into StoreQueryUtils, but I've been thinking that, rather than having a universal method to handleBasicQueries on any state store, the code will be cleaner with less type-checks and casts if you instead add a new handleWindowQueries and handleSessionQueries.

What do you think?

I was looking at your merged PR and coded the store query routines in a similar spirit. For instance, I follow your practice in the MeteredKeyValue store and applied it to MeteredWindow/Session stores. Similarly for the InMemoryKeyValue store I noticed you handle range queries in place, so I did the same for the InMemoryWindowStore. Generally if you feel we should handle queries in the util class (as some are) then probably we should be consistent across the stores and remove the local handlers and instead delegate to the store utils? Or alternatively pass the per store handlers to the StoreQueryUtils.

streams/src/main/java/org/apache/kafka/streams/state/internals/InMemorySessionStore.java

vvcephei · 2021-12-21T22:23:24Z

streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryWindowStore.java

+            if (windowRangeQuery.getTimeFrom().isPresent() &&
+                    windowRangeQuery.getTimeTo().isPresent()) {


It looks like this method will return "unknown query" if the window bounds aren't specified. Can we make sure there's a test case for WindowRangeQuery#withKey?

Ah, I was previously mistaken about this. My prior comment was made with your KIP-763 (https://cwiki.apache.org/confluence/display/KAFKA/KIP-763%3A+Range+queries+with+open+endpoints) in mind.

Upon looking at the WindowStore code and the KIP, I now see that we don't support open ranges on window stores, and that the KIP was only about KeyValue range queries.

I'm also just now catching on that you're using Optional.of there, not Optional.ofNullable, so it's always the case that we either have both bounds and no key or a key with no bounds.

I had a second concern that the window store is not handling the case of a range query with a key and no bounds, but upon examination of the WindowStore, I can see that there is no method to handle that case, so it makes sense to return "unknown query" for it (since the objective at this moment is parity). It might be nice if we additionally explain that the error is because the store can't handle the parameterization rather than the query itself, though.

And finally, just to clarify, I do think we should consolidate on handling the queries in StoreQueryUtils rather than in the stores themselves, so these comments are meant to apply to the handler in that class, and we should just delete the code in this class.

streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java

…Query.java Co-authored-by: John Roesler <vvcephei@users.noreply.github.com>

vvcephei · 2021-12-29T20:44:03Z

Hey @patrickstuedi , thanks for the last batch of updates. Aside from the build failure, I think these are my last comments:

After scrolling around the PR a few times to try and clarify whether we're handling all the right cases, I'm more convinced now that we really should consolidate all the query handling logic into StoreQueryUtils. I gather from your comment that you feel the same way. Let's just go for it in this PR. The thing you saw in the other PR was just an oversight.
Now that I'm really confronted with the reality of the code, I think it's going to be pretty confusing for people that WindowRangeQuery only works with WindowStore when you do withWindowStartRange(from, to) and it only works on SessionStore when you do withKey(key). I still think it's ok to do that, but I think we ought to add more description to the failure message than just "this store doesn't know how to handle this query". Can you replace the else cases there with QueryResult.forFailure(FailureReason.UNKNOWN_QUERY_TYPE, "...") and a message that more accurately explains what the problem is? It would also be nice to have negative test cases for these to verify the correct messages makes it to the user.

Once those points are addressed, we should be ready to merge!

vvcephei

Hey @patrickstuedi , I really hope you don't mind, but this PR has been dragging out for a long time, so I thought you might appreciate some help in getting it closed out.

I left a bunch of comments to explain the changes.

vvcephei · 2021-12-22T17:48:54Z

streams/src/main/java/org/apache/kafka/streams/query/WindowRangeQuery.java

+import java.time.Instant;
+import java.util.Optional;
+
+public class WindowRangeQuery<K, V> implements Query<KeyValueIterator<Windowed<K>, V>> {


Suggested change

public class WindowRangeQuery<K, V> implements Query<KeyValueIterator<Windowed<K>, V>> {

@Evolving

public class WindowRangeQuery<K, V> implements Query<KeyValueIterator<Windowed<K>, V>> {

vvcephei · 2021-12-28T03:29:14Z

streams/src/main/java/org/apache/kafka/streams/state/internals/InMemorySessionStore.java

@@ -317,13 +318,25 @@ public boolean isOpen() {
    @Override
    public <R> QueryResult<R> query(final Query<R> query, final PositionBound positionBound,
        final boolean collectExecutionInfo) {
+
+        if (query instanceof WindowRangeQuery) {


Hey @patrickstuedi , Github won't let me comment on the prior conversation, but thanks for pointing out that oversight in the in-memory key-value store! I went ahead and removed it as well.

vvcephei · 2021-12-29T19:53:43Z

streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryWindowStore.java

+            if (windowRangeQuery.getTimeFrom().isPresent() &&
+                    windowRangeQuery.getTimeTo().isPresent()) {


Ah, I was previously mistaken about this. My prior comment was made with your KIP-763 (https://cwiki.apache.org/confluence/display/KAFKA/KIP-763%3A+Range+queries+with+open+endpoints) in mind.

Upon looking at the WindowStore code and the KIP, I now see that we don't support open ranges on window stores, and that the KIP was only about KeyValue range queries.

I'm also just now catching on that you're using Optional.of there, not Optional.ofNullable, so it's always the case that we either have both bounds and no key or a key with no bounds.

I had a second concern that the window store is not handling the case of a range query with a key and no bounds, but upon examination of the WindowStore, I can see that there is no method to handle that case, so it makes sense to return "unknown query" for it (since the objective at this moment is parity). It might be nice if we additionally explain that the error is because the store can't handle the parameterization rather than the query itself, though.

And finally, just to clarify, I do think we should consolidate on handling the queries in StoreQueryUtils rather than in the stores themselves, so these comments are meant to apply to the handler in that class, and we should just delete the code in this class.

vvcephei · 2022-01-02T22:26:46Z

streams/src/main/java/org/apache/kafka/streams/query/QueryResult.java

@@ -58,7 +58,7 @@
        final Query<R> query,
        final StateStore store) {

-        return new FailedQueryResult<>(
+        return forFailure(


This just made it easier to inline the "unknown query" message.

vvcephei · 2022-01-02T22:27:30Z

streams/src/main/java/org/apache/kafka/streams/query/WindowKeyQuery.java

+    private WindowKeyQuery(final K key,
+                           final Optional<Instant> timeTo,
+                           final Optional<Instant> timeFrom) {


Since I was fixing stuff anyway, I went ahead and fixed a bunch of formatting issues that I didn't bother mentioning before.

vvcephei · 2022-01-02T22:57:30Z

streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java

+            Instant.ofEpochMilli(windowStart),
+            Instant.ofEpochMilli(windowStart),
+            extractor,
+            mkSet(2)


Since we're specifying the key, we expect only to get back windows with the value for that key. The aggregation we specified is to sum all values for the key, and it comes out to 2 because we only write one value for each key; namely, the value is the same number as the key.

vvcephei · 2022-01-02T22:59:56Z

streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java

+        // miss the window start range
+        shouldHandleWindowKeyQuery(
+            2,
+            Instant.ofEpochMilli(windowStart - 1),
+            Instant.ofEpochMilli(windowStart - 1),
+            extractor,
+            mkSet()
+        );
+
+        // miss the key
+        shouldHandleWindowKeyQuery(
+            999,
+            Instant.ofEpochMilli(windowStart),
+            Instant.ofEpochMilli(windowStart),
+            extractor,
+            mkSet()
+        );
+
+        // miss both
+        shouldHandleWindowKeyQuery(
+            999,
+            Instant.ofEpochMilli(windowStart - 1),
+            Instant.ofEpochMilli(windowStart - 1),
+            extractor,
+            mkSet()
        );


It seemed like a good idea to check a few other query configurations, but none of them showed any problems.

vvcephei · 2022-01-02T23:09:53Z

streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java

+                try (final WindowStoreIterator<V> iterator = queryResult.get(partition).getResult()) {
+                    while (iterator.hasNext()) {
+                        actualValue.add(valueExtactor.apply(iterator.next().value));
+                    }


These iterators need to be closed or they'll leak resources (it's the same for IQv1 as well).

vvcephei · 2022-01-02T23:10:27Z

streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java

+                try (final KeyValueIterator<Windowed<Integer>, V> iterator = queryResult.get(partition).getResult()) {
+                    while (iterator.hasNext()) {
+                        actualValue.add((Integer) iterator.next().value);
+                    }


Huh, looks like @vpapavas and I missed the need to close the iterator before.

vvcephei · 2022-01-02T23:12:47Z

streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java

        );
+
+        // Should fail to execute this query on a WindowStore.


I also added some cases to test the specialized failure messages.

vvcephei · 2022-01-03T04:15:57Z

The test failures were unrelated:

    Build / JDK 17 and Scala 2.13 / kafka.controller.ControllerIntegrationTest.testTopicIdUpgradeAfterReassigningPartitions()
    Build / JDK 11 and Scala 2.13 / kafka.admin.LeaderElectionCommandTest.[1] Type=Raft, Name=testTopicPartition, Security=PLAINTEXT
    Build / JDK 11 and Scala 2.13 / kafka.admin.LeaderElectionCommandTest.[1] Type=Raft, Name=testTopicPartition, Security=PLAINTEXT

patrickstuedi force-pushed the iqv2-session-query2 branch 23 times, most recently from fd081af to 9003a77 Compare December 14, 2021 14:44

patrickstuedi force-pushed the iqv2-session-query2 branch 5 times, most recently from 1416b91 to 88aa972 Compare December 21, 2021 15:07

vvcephei reviewed Dec 21, 2021

View reviewed changes

patrickstuedi force-pushed the iqv2-session-query2 branch from 20966d6 to cb77187 Compare December 22, 2021 12:31

patrickstuedi force-pushed the iqv2-session-query2 branch 3 times, most recently from bf4d070 to 05625f7 Compare December 22, 2021 13:19

Patrick Stuedi and others added 4 commits December 25, 2021 22:00

WindowKeyQuery and WindowRangeQuery

704def6

Update streams/src/main/java/org/apache/kafka/streams/query/WindowKey…

1a2efd0

…Query.java Co-authored-by: John Roesler <vvcephei@users.noreply.github.com>

Update streams/src/main/java/org/apache/kafka/streams/query/WindowKey…

4bc5f23

…Query.java Co-authored-by: John Roesler <vvcephei@users.noreply.github.com>

Updates to MeteredSessionStore

32ded30

patrickstuedi force-pushed the iqv2-session-query2 branch 2 times, most recently from c281005 to b4671ff Compare December 25, 2021 22:12

vvcephei mentioned this pull request Dec 28, 2021

MINOR: Remove duplicate IQv2 query handler #11630

Closed

3 tasks

vvcephei changed the title ~~WindowKeyQuery and WindowRangeQuery~~ KAFKA-13494: WindowKeyQuery and WindowRangeQuery Dec 28, 2021

vvcephei added the streams label Dec 28, 2021

Update InMemorySessionStore

fc613ff

patrickstuedi force-pushed the iqv2-session-query2 branch from b4671ff to fc613ff Compare December 31, 2021 14:13

vvcephei force-pushed the iqv2-session-query2 branch from 064b3ec to 5d0e5c5 Compare January 2, 2022 22:25

cleanup and fix

a193058

vvcephei force-pushed the iqv2-session-query2 branch from 5d0e5c5 to a193058 Compare January 2, 2022 23:11

vvcephei approved these changes Jan 2, 2022

View reviewed changes

vvcephei merged commit b8f1cf1 into apache:trunk Jan 3, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-13494: WindowKeyQuery and WindowRangeQuery #11567

KAFKA-13494: WindowKeyQuery and WindowRangeQuery #11567

patrickstuedi commented Dec 6, 2021

vvcephei left a comment

vvcephei Dec 21, 2021

patrickstuedi Dec 22, 2021 •

edited

vvcephei Dec 21, 2021

vvcephei Dec 29, 2021

vvcephei commented Dec 29, 2021

vvcephei left a comment

vvcephei Dec 22, 2021

vvcephei Dec 28, 2021 •

edited

vvcephei Dec 29, 2021

vvcephei Jan 2, 2022

vvcephei Jan 2, 2022

vvcephei Jan 2, 2022

vvcephei Jan 2, 2022

vvcephei Jan 2, 2022

vvcephei Jan 2, 2022

vvcephei Jan 2, 2022

vvcephei commented Jan 3, 2022

		if (windowRangeQuery.getTimeFrom().isPresent() &&
		windowRangeQuery.getTimeTo().isPresent()) {

	public class WindowRangeQuery<K, V> implements Query<KeyValueIterator<Windowed<K>, V>> {
	@Evolving
	public class WindowRangeQuery<K, V> implements Query<KeyValueIterator<Windowed<K>, V>> {

KAFKA-13494: WindowKeyQuery and WindowRangeQuery #11567

KAFKA-13494: WindowKeyQuery and WindowRangeQuery #11567

Conversation

patrickstuedi commented Dec 6, 2021

vvcephei left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

patrickstuedi Dec 22, 2021 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vvcephei commented Dec 29, 2021

vvcephei left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vvcephei Dec 28, 2021 • edited

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

vvcephei commented Jan 3, 2022

patrickstuedi Dec 22, 2021 •

edited

vvcephei Dec 28, 2021 •

edited