KAFKA-9929: Support reverse iterator on KeyValueStore #9137

jeqo · 2020-08-07T16:20:51Z

Implements KIP-617 on KeyValueStore.

Testing strategy: extend existing tests to validate reverse operations are supported.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

jeqo · 2020-08-08T09:25:17Z

cc @ableegoldman

This PR is ready for review, covering related feedback from #8976

ableegoldman

Thanks for splitting this up! Mostly minor comments on the KVStore PR, some general notes about the Streams style conventions:

always use braces for if/else/etc and put each condition on its own line
We're trying to move away from using @Test(expected) in the tests in favor of assertThrows. A lot of the tests haven't been migrated over yet, but just a heads up

ableegoldman · 2020-08-11T18:12:55Z

streams/src/main/java/org/apache/kafka/streams/state/ReadOnlyKeyValueStore.java

+     * Order is not guaranteed as bytes lexicographical ordering might not represent key order.
+     *
+     * @param from The first key that could be in the range, where iteration ends.
+     * @param to   The last key that could be in the range, where iteration starts from.


Seems a bit tricky to say that to is the variable where iteration starts from 😉 But I can see it both ways, so being clear in the javadocs is good enough for me

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

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

ableegoldman · 2020-08-11T20:21:59Z

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

+            iter.seek(from.get());
+            rawLastKey = to.get();
+        }
+        if (rawLastKey == null) {
            throw new NullPointerException("RocksDBRangeIterator: RawToKey is null for key " + to);


nit: RawToKey --> RawLastKey

Also it should be from for the reverse case, right?

right. I've moved this into each condition to set a correct exception message.

streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractKeyValueStoreTest.java

streams/src/test/java/org/apache/kafka/streams/state/internals/CachingKeyValueStoreTest.java

...c/test/java/org/apache/kafka/streams/state/internals/CompositeReadOnlyKeyValueStoreTest.java

jeqo · 2020-08-13T16:22:04Z

@ableegoldman this should be ready for another review :)

ableegoldman

Did a quick survey of everywhere that implements range() and all() to see if there might be anywhere else we need to add reverseRange and reverseAll. These guys seem to be missing their reverse counterparts:

KeyValueStoreReadWriteDecorator
KeyValueToTimestampedKeyValueByteStoreAdapter
Segment (this one is not strictly necessary until the WindowStore PR though)

The fact that you've had to add these to about 25 different classes already really highlights the mess of the store hierarchy 😬

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

ableegoldman · 2020-08-14T00:42:57Z

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

@@ -374,7 +412,7 @@ public Bytes peekNextKey() {
            if (next == null) {
                return allDone();
            } else {
-                if (comparator.compare(next.key.get(), upperBoundKey) <= 0) {
+                if (comparator.compare(next.key.get(), lastKey) <= 0) {


We need to branch on reverse here too, right?

Kind of unrelated, but WDYT about renaming RocksDBDualCFIterator to RocksDBDualCFAllIterator or something on the side? I feel like these iterators could be cleaned up a bit in general to be more understandable -- for example, it's weird that we do the iterator#seek-ing in the actual all() method but for range queries we do the seeking inside the iterator constructor.
Just thinking out loud though, we can do some followup refactoring once this is merged

Agree. I will continue the current approoach and create an issue to follow up this.

https://issues.apache.org/jira/browse/KAFKA-10409

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

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

streams/src/test/java/org/apache/kafka/streams/state/internals/CachingKeyValueStoreTest.java

ableegoldman · 2020-08-14T01:27:07Z

streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractKeyValueStoreTest.java

@@ -188,7 +188,55 @@ public void testPutGetRange() {
    }

    @Test
-    public void testPutGetRangeWithDefaultSerdes() {
+    public void testPutGetReverseRange() {


I think we should add some tests to RocksDBTimestampedStoreTest. I thought it would extend AbstractKeyValueStoreTest and thus benefit from everything you added here, but doesn't seem to be the case :/
Personally I found the RocksDBDualCFIterator logic a bit difficult to follow even before the reverse iteration, so it would be nice to have some tests specifically covering reverse iterators over multi-column-family timestamped stores

Just realized that, I also thought that path was tested. Good catch!

@ableegoldman I have extended RocksDBTimestampedStoreTest to use reverseAll and reverseRange as part of the current tests.
Unfortunately, AbstractKeyValueStoreTest tests do not fit with the creation path of Timestamped stores as pre inserted data is required.
Will add this to the same JIRA ticket to consider when refactoring iterators and tests.

Ah, that makes sense I guess. Looks like the new additions to RocksDBTimestampedStoreTest cover the cross-column family code path so that's good enough for now

ableegoldman

Thanks for the PR! Things LGTM now, I'll ping a committer to give this another pass and get it merged.

Let me know when the WindowStore PR is ready and I'll start reviewing the next one 😄

vvcephei

Thanks for the PR, @jeqo ! I made a partial pass today. It looks really good overall.

A high-level comment: in several places in this code, we have a internal boolean flag to represent the direction. It seems odd that in these places, true = reverse. It seems like the "natural" iteration order is forward, so we should have blocks like if (forward) /*forward logic*/ else /*reverse logic*/. But what we have here is if (not reverse) /*forward logic*/ else /*reverse logic*/. It's not exactly hard to understand, but still it seems to be creating unnecessary hoops for people to jump through when they are reading the code.

vvcephei · 2020-08-18T18:27:49Z

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

+            LOG.warn("Returning empty iterator for fetch with invalid key range: from > to. "
+                + "This may be due to serdes that don't preserve ordering when lexicographically comparing the serialized bytes. " +
+                "Note that the built-in numerical serdes do not follow this for negative numbers");


This warning seems to miss the most likely scenario, that the user just passed the arguments in the wrong order.

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

vvcephei

Hey @jeqo , I've completed my review. It looks good to me! Thanks so much for the KIP and PR.

Please let me know what you want to do about my prior comments, if anything :)

vvcephei · 2020-08-19T14:26:23Z

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

@@ -58,7 +62,7 @@ public synchronized boolean hasNext() {
            return allDone();
        } else {
            next = getKeyValue();
-            iter.next();
+            advanceIterator.accept(iter);


jeqo · 2020-08-20T09:32:00Z

@vvcephei thanks for your feedback! Inverting flag actually make things more readable. Changing to forward and if (forward) {} else {}.

vvcephei

Thanks @jeqo !

vvcephei · 2020-08-22T01:28:28Z

It looks like java 14 only failed while attempting to download some docs from Oracle, and java 8 failed on this unrelated test: org.apache.kafka.clients.admin.KafkaAdminClientTest.testMetadataRetries

Implements KIP-617 on WindowStore that depends on #9137. Testing strategy: extend existing tests to validate reverse operations are supported. Reviewers: A. Sophie Blee-Goldman <[email protected]>, Guozhang Wang <[email protected]>

This was referenced Aug 7, 2020

KAFKA-9929: Support backward iterator on WindowStore #9138

Merged

KIP-617: Allow Kafka Streams State Stores to be iterated backwards #8976

Closed

ableegoldman reviewed Aug 11, 2020

View reviewed changes

jeqo force-pushed the reverse-keyvaluestore branch from b4c44eb to 3a6db89 Compare August 13, 2020 12:17

ableegoldman reviewed Aug 14, 2020

View reviewed changes

ableegoldman reviewed Aug 17, 2020

View reviewed changes

vvcephei reviewed Aug 18, 2020

View reviewed changes

vvcephei reviewed Aug 19, 2020

View reviewed changes

jeqo added 11 commits August 20, 2020 10:21

key/value reverse operations

3f7f70b

refactor choose next key and value

07acb80

fix final params

58f6c19

improve ordering docs

e7ddbb9

fix range validator not needed

39d8e63

resolve nits

dddfa75

improve tests

834da34

improve tests

842c358

complete reverse paths

c634673

complete kvstore implementations

a8bdfd4

replace backward flag to forward flag

b124cb7

jeqo force-pushed the reverse-keyvaluestore branch from 218c678 to b124cb7 Compare August 20, 2020 09:21

improve range wrong order warning

9a0adeb

jeqo added 2 commits August 20, 2020 16:49

fix style

4dd4169

fix concat style

38c3bb5

vvcephei approved these changes Aug 22, 2020

View reviewed changes

vvcephei merged commit 89d0678 into apache:trunk Aug 22, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-9929: Support reverse iterator on KeyValueStore #9137

KAFKA-9929: Support reverse iterator on KeyValueStore #9137

jeqo commented Aug 7, 2020 •

edited

Loading

jeqo commented Aug 8, 2020

ableegoldman left a comment

ableegoldman Aug 11, 2020

ableegoldman Aug 11, 2020

ableegoldman Aug 11, 2020

jeqo Aug 12, 2020

jeqo commented Aug 13, 2020

ableegoldman left a comment

ableegoldman Aug 14, 2020

ableegoldman Aug 14, 2020

jeqo Aug 17, 2020

jeqo Aug 17, 2020

ableegoldman Aug 14, 2020

jeqo Aug 14, 2020

jeqo Aug 17, 2020

ableegoldman Aug 17, 2020

ableegoldman left a comment

vvcephei left a comment

vvcephei Aug 18, 2020

vvcephei left a comment

vvcephei Aug 19, 2020

jeqo commented Aug 20, 2020

vvcephei left a comment

vvcephei commented Aug 22, 2020

KAFKA-9929: Support reverse iterator on KeyValueStore #9137

KAFKA-9929: Support reverse iterator on KeyValueStore #9137

Conversation

jeqo commented Aug 7, 2020 • edited Loading

Committer Checklist (excluded from commit message)

jeqo commented Aug 8, 2020

ableegoldman 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

jeqo commented Aug 13, 2020

ableegoldman 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

ableegoldman left a comment

Choose a reason for hiding this comment

vvcephei left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vvcephei left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jeqo commented Aug 20, 2020

vvcephei left a comment

Choose a reason for hiding this comment

vvcephei commented Aug 22, 2020

jeqo commented Aug 7, 2020 •

edited

Loading