Add range and partition support on FoundationDB adapter for batch processing (#48)

Signed-off-by: Randy Hu <[email protected]>

Author: ruweih

src/main/java/org/janusgraph/diskstorage/foundationdb/FoundationDBKeyValueStore.java

@@ -15,6 +15,8 @@
 package org.janusgraph.diskstorage.foundationdb;
 
 import com.apple.foundationdb.KeyValue;
+import com.apple.foundationdb.LocalityUtil;
+import com.apple.foundationdb.async.CloseableAsyncIterator;
 import com.apple.foundationdb.async.AsyncIterator;
 import com.apple.foundationdb.directory.DirectorySubspace;
 import com.google.common.base.Preconditions;
@@ -30,14 +32,15 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.HashMap;
 
 /**
  * @author Ted Wilmes ([email protected])
  */
-public class FoundationDBKeyValueStore implements OrderedKeyValueStore {
+public class FoundationDBKeyValueStore implements OrderedKeyValueStore, AutoCloseable {
 
     private static final Logger log = LoggerFactory.getLogger(FoundationDBKeyValueStore.class);
 
@@ -254,6 +257,13 @@ public void delete(StaticBuffer key, StoreTransaction txh) throws BackendExcepti
             throw new PermanentBackendException(e);
         }
     }
+    public List<StaticBuffer> getBoundaryKeys() {
+        List<StaticBuffer> keys = new ArrayList<>();
+        try (CloseableAsyncIterator<byte[]> it = LocalityUtil.getBoundaryKeys(manager.db, db.range().begin, db.range().end)) {
+            it.forEachRemaining(key -> keys.add(getBuffer(db.unpack(key).getBytes(0))));
+        }
+        return keys;
+    }
 
     static StaticBuffer getBuffer(byte[] entry) {
         return new StaticArrayBuffer(entry);

src/main/java/org/janusgraph/diskstorage/foundationdb/FoundationDBRangeQuery.java

@@ -16,6 +16,8 @@
 
 import com.apple.foundationdb.KeySelector;
 import com.apple.foundationdb.subspace.Subspace;
+
+import org.janusgraph.diskstorage.StaticBuffer;
 import org.janusgraph.diskstorage.keycolumnvalue.keyvalue.KVQuery;
 
 /**
@@ -32,8 +34,13 @@ public FoundationDBRangeQuery(Subspace db, KVQuery kvQuery) {
         originalQuery = kvQuery;
         limit = kvQuery.getLimit();
 
-        byte[] startKey = db.pack(kvQuery.getStart().as(FoundationDBKeyValueStore.ENTRY_FACTORY));
-        byte[] endKey = db.pack(kvQuery.getEnd().as(FoundationDBKeyValueStore.ENTRY_FACTORY));
+        final StaticBuffer keyStart = kvQuery.getStart();
+        final StaticBuffer keyEnd = kvQuery.getEnd();
+
+        byte[] startKey = (keyStart == null) ?
+                db.range().begin : db.pack(keyStart.as(FoundationDBKeyValueStore.ENTRY_FACTORY));
+        byte[] endKey = (keyEnd == null) ?
+                db.range().end : db.pack(keyEnd.as(FoundationDBKeyValueStore.ENTRY_FACTORY));
 
         startKeySelector = KeySelector.firstGreaterOrEqual(startKey);
         endKeySelector = KeySelector.firstGreaterOrEqual(endKey);

src/main/java/org/janusgraph/diskstorage/foundationdb/FoundationDBStoreManager.java

@@ -56,7 +56,7 @@
  *
  * @author Ted Wilmes ([email protected])
  */
-public class FoundationDBStoreManager extends AbstractStoreManager implements OrderedKeyValueStoreManager {
+public class FoundationDBStoreManager extends AbstractStoreManager implements OrderedKeyValueStoreManager, AutoCloseable {
 
     private static final Logger log = LoggerFactory.getLogger(FoundationDBStoreManager.class);