Merge pull request #240 from amcp/0.1-titan_ids_fix

#228 Make ID store name configurable

Author: Alexander Patrikalakis

janusgraph-core/src/main/java/org/janusgraph/diskstorage/Backend.java

@@ -81,12 +81,13 @@ public class Backend implements LockerProvider, AutoCloseable {
      * <p/>
      * These names are fixed and should NEVER be changed. Changing these strings can
      * disrupt storage adapters that rely on these names for specific configurations.
+     * In the past, the store name for the ID table, janusgraph_ids, was also marked here,
+     * but to clear the upgrade path from Titan to JanusGraph, we had to pull it into
+     * configuration.
      */
     public static final String EDGESTORE_NAME = "edgestore";
     public static final String INDEXSTORE_NAME = "graphindex";
 
-    public static final String ID_STORE_NAME = "janusgraph_ids";
-
     public static final String METRICS_STOREMANAGER_NAME = "storeManager";
     public static final String METRICS_MERGED_STORE = "stores";
     public static final String METRICS_MERGED_CACHE = "caches";
@@ -103,12 +104,6 @@ public class Backend implements LockerProvider, AutoCloseable {
 
     public static final int THREAD_POOL_SIZE_SCALE_FACTOR = 2;
 
-    public static final Map<String, Integer> STATIC_KEY_LENGTHS = new HashMap<String, Integer>() {{
-        put(EDGESTORE_NAME, 8);
-        put(EDGESTORE_NAME + LOCK_STORE_SUFFIX, 8);
-        put(ID_STORE_NAME, 8);
-    }};
-
     private final KeyColumnValueStoreManager storeManager;
     private final KeyColumnValueStoreManager storeManagerLocking;
     private final StoreFeatures storeFeatures;
@@ -229,7 +224,7 @@ public Locker getLocker(String lockerName) {
     public void initialize(Configuration config) {
         try {
             //EdgeStore & VertexIndexStore
-            KeyColumnValueStore idStore = storeManager.openDatabase(ID_STORE_NAME);
+            KeyColumnValueStore idStore = storeManager.openDatabase(config.get(IDS_STORE_NAME));
 
             idAuthority = null;
             if (storeFeatures.isKeyConsistent()) {
@@ -414,7 +409,9 @@ public static KeyColumnValueStoreManager getStorageManager(Configuration storage
         StoreManager manager = getImplementationClass(storageConfig, storageConfig.get(STORAGE_BACKEND),
                 StandardStoreManager.getAllManagerClasses());
         if (manager instanceof OrderedKeyValueStoreManager) {
-            manager = new OrderedKeyValueStoreManagerAdapter((OrderedKeyValueStoreManager) manager, STATIC_KEY_LENGTHS);
+            manager = new OrderedKeyValueStoreManagerAdapter((OrderedKeyValueStoreManager) manager,
+                ImmutableMap.of(EDGESTORE_NAME, 8, EDGESTORE_NAME + LOCK_STORE_SUFFIX, 8,
+                    storageConfig.get(IDS_STORE_NAME), 8));
         }
         Preconditions.checkArgument(manager instanceof KeyColumnValueStoreManager,"Invalid storage manager: %s",manager.getClass());
         return (KeyColumnValueStoreManager) manager;

janusgraph-core/src/main/java/org/janusgraph/diskstorage/StaticBuffer.java

@@ -27,76 +27,62 @@
  */
 public interface StaticBuffer extends Comparable<StaticBuffer> {
 
-    public int length();
+    int length();
 
-    public byte getByte(int position);
+    byte getByte(int position);
 
-    public boolean getBoolean(int position);
+    boolean getBoolean(int position);
 
-    public short getShort(int position);
+    short getShort(int position);
 
-    public int getInt(int position);
+    int getInt(int position);
 
-    public long getLong(int position);
+    long getLong(int position);
 
-    public char getChar(int position);
+    char getChar(int position);
 
-    public float getFloat(int position);
+    float getFloat(int position);
 
-    public double getDouble(int position);
+    double getDouble(int position);
 
-    public byte[] getBytes(int position, int length);
+    byte[] getBytes(int position, int length);
 
-    public short[] getShorts(int position, int length);
+    short[] getShorts(int position, int length);
 
-    public int[] getInts(int position, int length);
+    int[] getInts(int position, int length);
 
-    public long[] getLongs(int position, int length);
+    long[] getLongs(int position, int length);
 
-    public char[] getChars(int position, int length);
+    char[] getChars(int position, int length);
 
-    public float[] getFloats(int position, int length);
+    float[] getFloats(int position, int length);
 
-    public double[] getDoubles(int position, int length);
+    double[] getDoubles(int position, int length);
 
-    public StaticBuffer subrange(int position, int length);
+    StaticBuffer subrange(int position, int length);
 
-    public StaticBuffer subrange(int position, int length, boolean invert);
+    StaticBuffer subrange(int position, int length, boolean invert);
 
-    public ReadBuffer asReadBuffer();
+    ReadBuffer asReadBuffer();
 
-    public<T> T as(Factory<T> factory);
+    <T> T as(Factory<T> factory);
 
     //Convenience method
-    public ByteBuffer asByteBuffer();
+    ByteBuffer asByteBuffer();
 
-    public interface Factory<T> {
+    interface Factory<T> {
 
-        public T get(byte[] array, int offset, int limit);
+        T get(byte[] array, int offset, int limit);
 
     }
 
-    public static final Factory<byte[]> ARRAY_FACTORY = new Factory<byte[]>() {
-        @Override
-        public byte[] get(byte[] array, int offset, int limit) {
-            if (offset==0 && limit==array.length) return array;
-            else return Arrays.copyOfRange(array,offset,limit);
-        }
-
+    Factory<byte[]> ARRAY_FACTORY = (array, offset, limit) -> {
+        if (offset==0 && limit==array.length) return array;
+        else return Arrays.copyOfRange(array,offset,limit);
     };
 
-    public static final Factory<ByteBuffer> BB_FACTORY = new Factory<ByteBuffer>() {
-        @Override
-        public ByteBuffer get(byte[] array, int offset, int limit) {
-            return ByteBuffer.wrap(array, offset, limit - offset);
-        }
-    };
+    Factory<ByteBuffer> BB_FACTORY = (array, offset, limit) -> ByteBuffer.wrap(array, offset, limit - offset);
 
-    public static final Factory<StaticBuffer> STATIC_FACTORY = new Factory<StaticBuffer>() {
-        @Override
-        public StaticBuffer get(byte[] array, int offset, int limit) {
-            return new StaticArrayBuffer(array, offset, limit);
-        }
-    };
+    Factory<StaticBuffer> STATIC_FACTORY = (array, offset, limit) -> new StaticArrayBuffer(array, offset, limit);
 
 }

janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/StoreFeatures.java

@@ -15,6 +15,7 @@
 package org.janusgraph.diskstorage.keycolumnvalue;
 
 import org.janusgraph.diskstorage.configuration.Configuration;
+import org.janusgraph.diskstorage.keycolumnvalue.scan.ScanJob;
 import org.janusgraph.diskstorage.util.time.TimestampProviders;
 
 /**
@@ -181,7 +182,7 @@ public interface StoreFeatures {
      *
      * Backends which don't support the notion of "local" strong consistency may
      * return the same configuration returned by
-     * {@link #getKeyConsistencyTxConfig()}.
+     * {@link #getKeyConsistentTxConfig()}.
      *
      * @return a locally (or globally) key-consistent tx config
      */
@@ -213,4 +214,5 @@ public interface StoreFeatures {
      * @return
      */
     boolean hasOptimisticLocking();
+
 }

janusgraph-core/src/main/java/org/janusgraph/graphdb/configuration/GraphDatabaseConfiguration.java

@@ -709,19 +709,29 @@ public boolean apply(@Nullable Integer integer) {
      * fraction of the id pool occupied and potentially lost. For write heavy applications, larger block sizes should
      * be chosen.
      */
-    public static final ConfigOption<Integer> IDS_BLOCK_SIZE = new ConfigOption<Integer>(IDS_NS,"block-size",
+    public static final ConfigOption<Integer> IDS_BLOCK_SIZE = new ConfigOption<>(IDS_NS,"block-size",
             "Globally reserve graph element IDs in chunks of this size.  Setting this too low will make commits " +
             "frequently block on slow reservation requests.  Setting it too high will result in IDs wasted when a " +
             "graph instance shuts down with reserved but mostly-unused blocks.",
             ConfigOption.Type.GLOBAL_OFFLINE, 10000);
 //    public static final String IDS_BLOCK_SIZE_KEY = "block-size";
 //    public static final int IDS_BLOCK_SIZE_DEFAULT = 10000;
 
+    /**
+     * The name of the ID store. Currently this defaults to janusgraph_ids. You can override the ID store to
+     * facilitate migration from JanusGraph's predecessor, Titan. Previously, this KCVStore was named titan_ids.
+     */
+    public static final ConfigOption<String> IDS_STORE_NAME = new ConfigOption<>(IDS_NS, "store-name",
+        "The name of the ID KCVStore. IDS_STORE_NAME is meant to be used only for backward compatibility with Titan, " +
+            "and should not be used explicitly in normal operations or in new graphs.",
+        ConfigOption.Type.GLOBAL_OFFLINE, "janusgraph_ids");
+
+
     /**
      * If flush ids is enabled, vertices and edges are assigned ids immediately upon creation. If not, then ids are only
      * assigned when the transaction is committed.
      */
-    public static final ConfigOption<Boolean> IDS_FLUSH = new ConfigOption<Boolean>(IDS_NS,"flush",
+    public static final ConfigOption<Boolean> IDS_FLUSH = new ConfigOption<>(IDS_NS,"flush",
             "When true, vertices and edges are assigned IDs immediately upon creation.  When false, " +
             "IDs are assigned only when the transaction commits. Must be disabled for graph partitioning to work.",
             ConfigOption.Type.MASKABLE, true);
@@ -744,7 +754,7 @@ public boolean apply(@Nullable Integer integer) {
      * of the current block is consumed, a new block will be allocated. Larger values should be used if a lot of ids
      * are allocated in a short amount of time. Value must be in (0,1].
      */
-    public static final ConfigOption<Double> IDS_RENEW_BUFFER_PERCENTAGE = new ConfigOption<Double>(IDS_NS,"renew-percentage",
+    public static final ConfigOption<Double> IDS_RENEW_BUFFER_PERCENTAGE = new ConfigOption<>(IDS_NS,"renew-percentage",
             "When the most-recently-reserved ID block has only this percentage of its total IDs remaining " +
             "(expressed as a value between 0 and 1), JanusGraph asynchronously begins reserving another block. " +
             "This helps avoid transaction commits waiting on ID reservation even if the block size is relatively small.",

janusgraph-hadoop-parent/janusgraph-hadoop-core/pom.xml

@@ -35,6 +35,11 @@
             <version>${project.version}</version>
             <!-- TODO this should be scoped test -->
         </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>janusgraph-hbase-core</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         <dependency>
             <groupId>${project.groupId}</groupId>
             <artifactId>janusgraph-cassandra</artifactId>
@@ -84,9 +89,6 @@
                             <goal>testCompile</goal>
                             -->
                         </goals>
-                        <configuration>
-                            <providerSelection>1.8</providerSelection>
-                        </configuration>
                     </execution>
                 </executions>
             </plugin>

janusgraph-hadoop-parent/janusgraph-hadoop-core/src/main/java/org/janusgraph/hadoop/formats/hbase/HBaseBinaryInputFormat.java

@@ -42,13 +42,15 @@
 import java.lang.reflect.Method;
 import java.util.List;
 
+import com.google.common.collect.BiMap;
+
 public class HBaseBinaryInputFormat extends AbstractBinaryInputFormat {
 
     private static final Logger log = LoggerFactory.getLogger(HBaseBinaryInputFormat.class);
 
     private final TableInputFormat tableInputFormat = new TableInputFormat();
     private RecordReader<ImmutableBytesWritable, Result> tableReader;
-    private byte[] inputCFBytes;
+    private byte[] edgeStoreFamily;
     private RecordReader<StaticBuffer, Iterable<Entry>> janusgraphRecordReader;
 
     @Override
@@ -60,7 +62,7 @@ public List<InputSplit> getSplits(final JobContext jobContext) throws IOExceptio
     public RecordReader<StaticBuffer, Iterable<Entry>> createRecordReader(final InputSplit inputSplit, final TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
         tableReader = tableInputFormat.createRecordReader(inputSplit, taskAttemptContext);
         janusgraphRecordReader =
-                new HBaseBinaryRecordReader(tableReader, inputCFBytes);
+                new HBaseBinaryRecordReader(tableReader, edgeStoreFamily);
         return janusgraphRecordReader;
     }
 
@@ -82,13 +84,14 @@ public void setConf(final Configuration config) {
         // TODO the space-saving short name mapping leaks from HBaseStoreManager here
         if (janusgraphConf.get(HBaseStoreManager.SHORT_CF_NAMES)) {
             try {
-                cfName = HBaseStoreManager.shortenCfName(cfName);
+                final BiMap<String,String> shortCfMap = HBaseStoreManager.createShortCfMap(janusgraphConf);
+                cfName = HBaseStoreManager.shortenCfName(shortCfMap, cfName);
             } catch (PermanentBackendException e) {
                 throw new RuntimeException(e);
             }
         }
         scanner.addFamily(cfName.getBytes());
-        inputCFBytes = Bytes.toBytes(cfName);
+        edgeStoreFamily = Bytes.toBytes(cfName);
 
         //scanner.setFilter(getColumnFilter(janusgraphSetup.inputSlice(this.vertexQuery))); // TODO
         //TODO (minor): should we set other options in http://hbase.apache.org/apidocs/org/apache/hadoop/hbase/client/Scan.html for optimization?
@@ -109,7 +112,7 @@ public RecordReader<ImmutableBytesWritable, Result> getTableReader() {
     }
 
     public byte[] getEdgeStoreFamily() {
-        return inputCFBytes;
+        return edgeStoreFamily;
     }
 
     private Filter getColumnFilter(SliceQuery query) {