async iterator final submission

[jlazdw]

This PR is to introduce asynchronous Iterator for getSlice And getSlices methods in FoundationDBKeyvalueStore.java ,to greatly leverage FoundationDB’s Asynchronous Iterator for on-demand data stream pulling, better memory efficiency, better parallelism for MultiQuery support. The asynchronous iterator relies on the FDB java library to take care of Completable Futures, rather than introducing additional Completable Futures in this storage plugin.

To support this new Async Iterator mode, we introduce “iterator” mode in FoundationDBConfigOptions.java. The existing getSlice and getSlices() is in the new (default) mode called “list” mode, as the query result from FDB java library gets converted to List-based result immediately.

We also provide the fix on getSlices in the current implementation.

[linux-foundation-easycla]

The committers are authorized under a signed CLA.

• ✅ jlazdw (85b8f74, ba55af7)

[rngcntr]

That sound like a very good improvement to me, thanks @jlazdw!
I'm kindly asking @ruweih to help me out reviewing it, because he is also working on an asynchonous implementation of getSlice.

[jackson-chris]

+1 LGTM. Thank you so much for this contribution @jlazdw

[ruweih]

Seems we can extend from existing FoundationDBRecordIterator with a new constructor accepting iterator of KeyValue instead of list:
https://github.com/JanusGraph/janusgraph-foundationdb/blob/e8cf024519045f501a7ad83b8090c61e9ebd82b3/src/main/java/org/janusgraph/diskstorage/foundationdb/FoundationDBRecordIterator.java

In the constructor of this class, we can leverage the Guava's iterator utility to make the code cleaner:

            Iterator<KeyValue> entries = Iterators.filter(
                Iterators.transform(result, keyValue -> {
                    StaticBuffer key = getBuffer(db.unpack(keyValue.getKey()).getBytes(0));
                    if (selector.include(key)) {
                        return new KeyValueEntry(key, getBuffer(keyValue.getValue()));
                    } else {
                        return null;
                    }
                }),
                keyValue -> keyValue != null);
        }
       super(entries);

We can then reuse the base class implementation on iterator interface

[jlazdw]

@ruweih: I tried to incorporate your suggestion to the code. and I just found that the code that you suggested above can not be compiled, as Iterator entries has the type "KeyValue" and the inner transformer returns the type "KeyValueEntry". The type "KeyValue" and "KeyValueEntry" has no inheritance relationship.

The current class "FoundationDBRecordIterator.java" has the private member:

private final Iterator entries

It is at the method of "next" that the "KeyValueEntry" gets constructed on the fly. The Selector logic is applied at "FoudationDBKeyValueStore" getSlice method, before the FoundationDBRecordIterator object is constructed. https://github.com/JanusGraph/janusgraph-foundationdb/blob/master/src/main/java/org/janusgraph/diskstorage/foundationdb/FoundationDBKeyValueStore.java#L134.

Therefore, if we want to keep the current FoundationDBRecordIterator's interface of using "KeyValue" and the current implementation in FoundationDBKeyValueStore's getSlice method, it is better to keep the "KeyValueEntry" construction only at the "next" method for FoundationDBRecordIteratorForAsync, just like FoundationDBRecordIterator's current implementation, instead of having the KeyValueEntry constructed at the constructor time.

Furthermore, unlike the synchronous iterator (that is, the current FoundationDBRecordIterator.java), our asynchronous iterator will need to re-start the FDB transaction if the transaction is too old, under "read_commit_with_write" mode, this is shown in the fetchNext() method of FoundationDBRecordIteratorAsync.java. As a result, for a long query, we need to construct many times, the re-newed version of "entries" with type of AsyncIterator. Therefore, the Guava's Iterator based code will have to be invoked not just in the constructor but also in the fetchNext() method, which makes the code sort of duplicated.

If you have better way to refactor the code for the Record Iterator and the FoundationDBKeyvalueStore, please provide more details and I will incorporate your suggestion to the code.

[rngcntr]

I agree with @ruweih that it is a little bit odd to have two completely different iterators for synchronous and asynchronous results, but I see your point and understand why it is so hard to keep the current implementation's interface. Would it sound easier and more convenient to you incorporating the FoundationDBRecordIterator's functionality into the new FoundationDBRecordIteratorForAsync?
In my opinion, it is perfectly fine to change the way the synchronous iterator works by incorporating the KeySelector into the iterator for example.
In the end, changing the interface in order to reduce the horizontal complexity is a design decision that sounds plausible to me.

[jlazdw]

@rngcntr: thanks for the suggestion on having KeySelector to be pushed down to both synchronous iterator and the asynchronous iterator. Let me try to make the corresponding changes and re-submit the PR.

[jlazdw]

@rngcntr: with your suggestion above, I refactoring FoundationDBRecordIterator and FoundationDBRecordAsyncIterator, to have FoundationDBRecordAysncIterator to inherit from FoundationDBRecordIterator. The hasNext() and next() logic is in the base class of FoundationDBRecordIterator. The difference is on the fetchNext() logic. The async iterator requires possible transaction re-start in fetchNext() logic.

The Selector logic is implemented in the base class, FoundationDBRecordIterator. It exposes the cursor "fetched", that allows the async iterator to set the "skip" offset when the transaction gets re-started and the getRangeIter() gets invoked again.

[rngcntr]

Sounds like a well designed solution to me! I'll most probably have another look at it later this week.

[ruweih]

@jlazdw The code snippet in my previous post was from my POC of async iterator on original 0.1.0 release code, which seems having difference signature: https://github.com/JanusGraph/janusgraph-foundationdb/blob/v0.1.0/src/main/java/com/experoinc/janusgraph/diskstorage/foundationdb/FoundationDBKeyValueStore.java#L148

With the current code base, it could be further simplified to just return KeyValue:

            Iterator<KeyValue> entries = Iterators.filter(
                Iterators.transform(result, keyValue -> {
                    StaticBuffer key = getBuffer(db.unpack(keyValue.getKey()).getBytes(0));
                    if (selector.include(key)) {
                        return keyValue;
                    } else {
                        return null;
                    }
                }),
                keyValue -> keyValue != null);
        }
       super(entries);

The transaction restart logic could happen inside the inner iterators mapping method as well. I noticed there is new changes been checked in , this might not necessary as long as the common iteration logic been reused.

[jlazdw]

@ruweih: the difficulty that I run into to try to use the above Iterators filter/transform that you suggested is that I need to record the cursor "fetched", so that when the transaction gets restarted in the Async iterator (inherited from the FoundationDBRecordIterator), I can skip the entries that have been already fetched. This state information will have to be recorded at the iterator level. https://github.com/jlazdw/janusgraph-foundationdb/blob/Async_iterator_final_submission/src/main/java/org/janusgraph/diskstorage/foundationdb/FoundationDBRecordIterator.java#L60

Also, when the transaction gets restarted, the iterator will need to be reset in terms of entries, but not the cursor "fetched", https://github.com/jlazdw/janusgraph-foundationdb/blob/Async_iterator_final_submission/src/main/java/org/janusgraph/diskstorage/foundationdb/FoundationDBRecordAsyncIterator.java#L80.

Could you help come up with the way to incorporate the state "fetched" tracking using the filter/transform, and can be reset on-the-fly later when transaction gets restarted, if that is possible?

[rngcntr]

Thanks for adding so many log outputs everywhere, that will certainly help us out hunting bugs in the future!

I've added a few comments, most of them requesting minor changes. The reason why I don't approve this PR yet is that almost none of the changes are covered in the existing tests. I think to ensure the integrity of the async methods and the iterator, we should also run the existing tests in iterator mode.

[rngcntr]

I agree with @ruweih that it is a little bit odd to have two completely different iterators for synchronous and asynchronous results, but I see your point and understand why it is so hard to keep the current implementation's interface. Would it sound easier and more convenient to you incorporating the FoundationDBRecordIterator's functionality into the new FoundationDBRecordIteratorForAsync?
In my opinion, it is perfectly fine to change the way the synchronous iterator works by incorporating the KeySelector into the iterator for example.
In the end, changing the interface in order to reduce the horizontal complexity is a design decision that sounds plausible to me.

[rngcntr]

Looks much cleaner now, thank you @jlazdw!
Only two more suggestions that came to my mind. I hope you are not annoyed yet 😄

[rngcntr]

The implementation of next() does not match the expectation of what an Iterator does:
https://docs.oracle.com/javase/8/docs/api/java/util/Iterator.html#next--

Repeated calls of next() without an intermediate hasNext() should still continue the iteration. In other words, next() should not rely on hasNext() being called.
In addition, calling next() on an exhausted iteration should result in a NoSuchElementException, not null.

[jlazdw]

@rngcntr: thanks for the suggestion on the iterator interface. I have made some changes following your suggestion. Over the last two days when I was making a proposal on TransactionContext to the JanusGraph developer community, JanusGraph/janusgraph#2238, I realize that for Threaded Transactions, the current Async Iterator will need to address multiple threads to re-set the same transaction. So I will make further changes in the next several days to handle the threaded transaction.

I have two related questions: (1) why in the current master branch, the test class of FoundationDBGraphConcurrentTest gets "disabled", https://github.com/JanusGraph/janusgraph-foundationdb/blob/master/src/test/java/org/janusgraph/graphdb/foundationdb/FoundationDBGraphConcurrentTest.java#L29
(2) The related JanusGraphConcurrentTest's test cases only creates a Transaction object in each worker thread launched, it is not Threaded Transaction at all, based on the "threaded transaction" definition described at: https://docs.janusgraph.org/basics/transactions/.

In fact, there is no test in the current JanusGraph test suites to test out threaded transactions, by searching the method invocation of "createThreadedTx()" in the entire test JanusGraph test suite.

That brings the question on: whether the current JanusGraph really support "threaded transaction" or not?

[rngcntr]

Regarding your first question: The test was already failing when we took over the project and until now, nobody started to investigate why. There's an open issue to resolve the failing test case: #28

For the second question I get your point but I have no Idea about JanusGraph's transaction handling and I feel unable to give any qualified answer to that. If you see a serious problem here, you should maybe report it to the janusgraph-dev mailing list.

[farodin91]

@jlazdw Do you want to squash all commits? Afterwards we should be able to merge it.

[jlazdw]

@farodin91: I have squashed all the commits up to the merge point. In between my commits, there was a PR merge to master. And thus the other commit 086b8ab separated my commits. Please let me know if you like me to squash further or the current squash is OK. If it is OK, please merge the current PR to the master. I am working on improving the async iterator further (including enabling the FoundationDBGraphConcurrentTest related tests that have been disabled in current master). I will try to make another PR in the next two weeks.

[farodin91]

@jlazdw In the normal case, we just want to have one commit in top of the current it possible. It would be great.

[farodin91]

I am working on improving the async iterator further (including enabling the FoundationDBGraphConcurrentTest related tests that have been disabled in current master). I will try to make another PR in the next two weeks.

Sounds awesome

[jerryjch]

@farodin91 I think we can also use the "Squash and Merge' option to merge this PR. There is no conflict to merge.

[jlazdw]

@farodin91: I talked to Jerry (@jerryjch ) and he suggested above on "Squash and Merge" from your side when you merge the PR to the master. If you perform squash, could you use the following last commit message from my submission branch: jlazdw@ec1ca28, which is:

"
introduce asynchronous iterator
unify the interface for synchronous iterator and asynchronous iterator
minor clean-up on code for logging and class import and refactor record iterators (sync and async)

minor changes to the code:
*re-defining enum for sync and asyn iterator
*backend exception propagation clean-up

made changes to test environment and README.md:
*introduce environment-controlled option to use iterator or list mode for testing
*add -Dgetrangemode=iterator and -Dgetrangemode=list to Travis CI jobs
*add section in README.md to explain iterator/list mode choice for GetRange related queries

"

[farodin91]

Thank you!