Reduced travel graph scenario test to minimal code.

Author: Alexander Patrikalakis

.gitignore

@@ -16,5 +16,3 @@
 /.settings
 .DS_Store
 *~
-.*~
-!/src/test/java/com/amazon/janusgraph/diskstorage/dynamodb/

pom.xml

@@ -66,6 +66,13 @@
             <name>Justin Panian</name>
             <email>[email protected]</email>
         </contributor>
+        <contributor>
+            <name>Addison Slabaugh</name>
+            <email>[email protected]</email>
+        </contributor>
+        <contributor>
+            <name>John Stephenson</name>
+        </contributor>
         <contributor>
             <name>Johan Jacobs</name>
             <email>[email protected]</email>

src/test/java/com/amazon/janusgraph/AbstractGraph.java

@@ -14,10 +14,20 @@
  */
 package com.amazon.janusgraph;
 
+import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.inV;
 import static org.junit.Assert.fail;
 
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.net.URL;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
+import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
 import org.apache.commons.configuration.Configuration;
@@ -28,10 +38,13 @@
 import org.janusgraph.core.Cardinality;
 import org.janusgraph.core.JanusGraph;
 import org.janusgraph.core.JanusGraphFactory;
+import org.janusgraph.core.JanusGraphTransaction;
+import org.janusgraph.core.Multiplicity;
 import org.janusgraph.core.PropertyKey;
 import org.janusgraph.core.VertexLabel;
 import org.janusgraph.core.schema.ConsistencyModifier;
 import org.janusgraph.core.schema.JanusGraphIndex;
+import org.janusgraph.core.schema.JanusGraphManagement;
 import org.janusgraph.core.schema.SchemaAction;
 import org.janusgraph.core.schema.SchemaStatus;
 import org.janusgraph.diskstorage.BackendException;
@@ -41,34 +54,45 @@
 import org.janusgraph.graphdb.database.management.ManagementSystem;
 import org.janusgraph.graphdb.transaction.StandardJanusGraphTx;
 import org.junit.Test;
+import org.junit.experimental.categories.Category;
 
 import com.amazon.janusgraph.diskstorage.dynamodb.BackendDataModel;
 import com.amazon.janusgraph.diskstorage.dynamodb.DynamoDBStoreTransaction;
+import com.amazon.janusgraph.testcategory.IsolateRemainingTestsCategory;
+import com.google.common.base.Preconditions;
 import com.google.common.base.Stopwatch;
 
+import lombok.Getter;
+import lombok.ToString;
+import lombok.extern.slf4j.Slf4j;
+
 /**
  *
  * @author Alexander Patrikalakis
+ * @author John Stephenson
+ * @author Addison Slabaugh
  *
  */
+@Slf4j
+@Category({IsolateRemainingTestsCategory.class})
 public class ScenarioTests {
 
-    public static final String LABEL = "myLabel";
-    public static final String BY_DATABASE_METADATA_VERSION = "byDatabaseMetadataVersion";
-    public static final String DATABASE_METADATA_LABEL = "databaseMetadata";
-    public static final String VERSION_PROPERTY = "version";
+    private static final String LABEL = "myLabel";
+    private static final String BY_DATABASE_METADATA_VERSION = "byDatabaseMetadataVersion";
+    private static final String DATABASE_METADATA_LABEL = "databaseMetadata";
+    private static final String VERSION_PROPERTY = "version";
     private static final boolean USE_STORAGE_NATIVE_LOCKING = true;
     private static final boolean USE_JANUSGRAPH_LOCKING = false;
     private static final boolean USE_GRAPHINDEX_LOCKING = true;
     private static final boolean USE_EDGESTORE_LOCKING = true;
-    public static final String VERSION_ONE = "0.0.1";
-    public static final String VERSION_TWO = "0.0.2";
-    public static final String VERSION_THREE = "0.0.3";
+    private static final String VERSION_ONE = "0.0.1";
+    private static final String VERSION_TWO = "0.0.2";
+    private static final String VERSION_THREE = "0.0.3";
 
     /**
      * This test is to demonstrate performance in response to a report of elevated latency for committing 30 vertices.
      * http://stackoverflow.com/questions/42899388/titan-dynamodb-local-incredibly-slow-8s-commit-for-30-vertices
-     * @throws BackendException
+     * @throws BackendException in case cleanUpTables fails
      */
     @Test
     public void performanceTest() throws BackendException {
@@ -92,12 +116,137 @@ public void lockingTest() throws BackendException, InterruptedException {
 
     /**
      * https://stackoverflow.com/questions/44535054/generalizing-dynamodb-janusgraph-factory-lock-and-schema-problems
-     * @throws Exception
      */
+    public enum Relationship {
+        instanceOf, hotelBrandType;
+    }
+    @Getter
+    @ToString
+    public class Triple {
+        private final String leftPropertyValue;
+        private final String leftPropertyName;
+        private final Relationship relationship;
+        private final String rightPropertyValue;
+        private final String rightPropertyName;
+
+        Triple(final String[] line) {
+            Preconditions.checkArgument(line.length == 3);
+            final String[] left = line[0].split(":");
+            final String[] right = line[2].split(":");
+            this.leftPropertyName = left[0];
+            this.leftPropertyValue = left[1];
+            this.relationship = Relationship.valueOf(line[1]);
+            this.rightPropertyName = right[0];
+            this.rightPropertyValue = right[1];
+        }
+    }
+    private static void createHotelSchema(JanusGraph graph) {
+        //another issue, you should only try to create the schema once.
+        //you use uniqueness constraints, so you need to define the schema up front with the unique() call,
+        //but if you did not use uniqueness constraints, you could just let JanusGraph create the schema for you.
+        final JanusGraphManagement mgmt = graph.openManagement();
+        final PropertyKey brandtypePropertyKey = mgmt.makePropertyKey("brandtype").dataType(String.class).make();
+        mgmt.buildIndex("brandtypeIndex", Vertex.class).addKey(brandtypePropertyKey).unique().buildCompositeIndex();
+        final PropertyKey namePropertyKey = mgmt.makePropertyKey("name").dataType(String.class).make();
+        mgmt.buildIndex("nameIndex", Vertex.class).addKey(namePropertyKey).unique().buildCompositeIndex();
+        mgmt.makeEdgeLabel(Relationship.hotelBrandType.name()).multiplicity(Multiplicity.MANY2ONE).make();
+        mgmt.makeEdgeLabel(Relationship.instanceOf.name()).multiplicity(Multiplicity.MANY2ONE).make();
+        mgmt.commit();
+    }
+
+    private void tripleIngestBase(BiConsumer<StandardJanusGraph, List<Triple>> writer) throws BackendException {
+        final Stopwatch watch = Stopwatch.createStarted();
+        final StandardJanusGraph graph = (StandardJanusGraph) JanusGraphFactory.open(TestGraphUtil.instance.createTestGraphConfig(BackendDataModel.MULTI));
+        log.info("Created graph in " + watch.elapsed(TimeUnit.MILLISECONDS) + " ms");
+        watch.reset();
+        watch.start();
+        createHotelSchema(graph);
+        log.info("Created schema in " + watch.elapsed(TimeUnit.MILLISECONDS) + " ms");
+
+        watch.reset();
+        watch.start();
+        final URL url = ScenarioTests.class.getClassLoader().getResource("META-INF/HotelTriples.txt");
+        final List<Triple> lines;
+        try (final BufferedReader bf = new BufferedReader(new InputStreamReader(url.openStream()))) {
+            lines = bf.lines()
+                .map(line -> line.split("\t"))
+                .map(Triple::new)
+                .collect(Collectors.toList());
+        } catch (IOException e) {
+            throw new IllegalStateException("Error processing triple file", e);
+        }
+        log.info("Read file into Triple objects in " + watch.elapsed(TimeUnit.MILLISECONDS) + " ms");
+        watch.reset();
+        watch.start();
+        writer.accept(graph, lines);
+        log.info("Added objects in " + watch.elapsed(TimeUnit.MILLISECONDS) + " ms");
+        TestGraphUtil.instance.cleanUpTables();
+    }
+
+    @Test
+    public void processTripleWithTraversals() throws BackendException {
+        tripleIngestBase((StandardJanusGraph graph, List<Triple> lines) -> {
+            final GraphTraversalSource g = graph.traversal();
+            lines.parallelStream().forEach(triple -> {
+                final Vertex left = getVertexIfDoesntExist(g, triple.getLeftPropertyName(),
+                    triple.getLeftPropertyValue());
+                final Vertex right = getVertexIfDoesntExist(g, triple.getRightPropertyName(),
+                    triple.getRightPropertyValue());
+                //your original method was creating vertices in the processRelationship method.
+                //this caused the uniqueness constraint violation (one of a few issues in your
+                //original code) because you have a unique index on the rightPropertyName
+                g.V()
+                    .is(left)
+                    .outE(triple.getRelationship().name())
+                    .filter(inV().is(right))
+                    .tryNext()
+                    .orElseGet(() -> left.addEdge(triple.getRelationship().name(), right));
+            });
+            final Stopwatch watch = Stopwatch.createStarted();
+            g.tx().commit();
+            log.info("Committed in " + watch.elapsed(TimeUnit.MILLISECONDS) + " ms");
+            watch.stop();
+        });
+    }
+    private static Vertex getVertexIfDoesntExist(GraphTraversalSource g,
+        String propertyName, String propertyValue) {
+        return g.V()
+            .has(propertyName, propertyValue)
+            .tryNext()
+            .orElseGet(() -> g.addV().property(propertyName, propertyValue).next());
+    }
+
     @Test
-    public void tripleIngestTest(StandardJanusGraph graph) throws Exception {
-        TravelGraphFactory graphFactory = new TravelGraphFactory();
-        graphFactory.loadGraphFactory(graph, false);
+    public void processTripleWithMaps() throws BackendException {
+        final ConcurrentMap<String, Vertex> brandTypeMap = new ConcurrentHashMap<>();
+        final ConcurrentMap<String, Vertex> companyMap = new ConcurrentHashMap<>();
+        final ConcurrentMap<String, Vertex> hotelBrandMap = new ConcurrentHashMap<>();
+        tripleIngestBase((StandardJanusGraph graph, List<Triple> triples) -> {
+            final JanusGraphTransaction threadedGraph = graph.newTransaction();
+            triples.parallelStream().forEach(triple -> {
+                final Vertex outV = hotelBrandMap.computeIfAbsent(triple.getLeftPropertyValue(),
+                    value -> threadedGraph.addVertex(triple.getLeftPropertyName(), value));
+                //your original method was creating vertices in the processRelationship method.
+                //this caused the uniqueness constraint violation (one of a few issues in your
+                //original code) because you have a unique index on the rightPropertyName
+                switch (triple.getRelationship()) {
+                    case hotelBrandType:
+                        outV.addEdge("hotelBrandType",brandTypeMap.computeIfAbsent(triple.getRightPropertyValue(),
+                            value -> threadedGraph.addVertex(triple.getRightPropertyName(), value)));
+                        break;
+                    case instanceOf:
+                        outV.addEdge("instanceOf", companyMap.computeIfAbsent(triple.getRightPropertyValue(),
+                            value -> threadedGraph.addVertex(triple.getRightPropertyName(), value)));
+                        break;
+                    default:
+                        throw new IllegalArgumentException("unexpected relationship type");
+                }
+            });
+            final Stopwatch watch = Stopwatch.createStarted();
+            threadedGraph.commit();
+            log.info("Committed in " + watch.elapsed(TimeUnit.MILLISECONDS) + " ms");
+            watch.stop();
+        });
     }
 
     @Test
@@ -117,7 +266,7 @@ private void createSchemaAndDemoLockExpiry(boolean useNativeLocking, boolean use
         }
     }
 
-    private static final DynamoDBStoreTransaction getStoreTransaction(ManagementSystem mgmt) {
+    private static DynamoDBStoreTransaction getStoreTransaction(ManagementSystem mgmt) {
         return DynamoDBStoreTransaction.getTx(((CacheTransaction) mgmt.getWrappedTx().getTxHandle().getStoreTransaction()).getWrappedTransaction());
     }
 
@@ -193,7 +342,7 @@ private JanusGraph createGraphWithSchema(final boolean useNativeLocking, final b
         final JanusGraphIndex indexAfterFirstCommit = mgmt.getGraphIndex(BY_DATABASE_METADATA_VERSION);
         final PropertyKey propertyKeySecond = mgmt.getPropertyKey(VERSION_PROPERTY);
         if (indexAfterFirstCommit.getIndexStatus(propertyKeySecond) == SchemaStatus.INSTALLED) {
-            ((ManagementSystem) mgmt).awaitGraphIndexStatus(graph, BY_DATABASE_METADATA_VERSION).status(SchemaStatus.REGISTERED).timeout(10, java.time.temporal.ChronoUnit.MINUTES).call();
+            ManagementSystem.awaitGraphIndexStatus(graph, BY_DATABASE_METADATA_VERSION).status(SchemaStatus.REGISTERED).timeout(10, java.time.temporal.ChronoUnit.MINUTES).call();
         }
         mgmt.commit();
 
@@ -216,15 +365,15 @@ private JanusGraph createGraphWithSchema(final boolean useNativeLocking, final b
             if(useNativeLocking) {
                 System.out.println("mgmt tx five " + getStoreTransaction(mgmt).getId());
             }
-            ((ManagementSystem) mgmt).awaitGraphIndexStatus(graph, BY_DATABASE_METADATA_VERSION).status(SchemaStatus.ENABLED).timeout(10, java.time.temporal.ChronoUnit.MINUTES)
+            ManagementSystem.awaitGraphIndexStatus(graph, BY_DATABASE_METADATA_VERSION).status(SchemaStatus.ENABLED).timeout(10, java.time.temporal.ChronoUnit.MINUTES)
                 .call();
         }
         mgmt.commit();
         //END code from second code listing (simplified index setup)
         return graph;
     }
 
-    private static final DynamoDBStoreTransaction getTxFromGraph(StandardJanusGraph graph) {
+    private static DynamoDBStoreTransaction getTxFromGraph(StandardJanusGraph graph) {
         return DynamoDBStoreTransaction.getTx(((CacheTransaction) ((StandardJanusGraphTx) graph.getCurrentThreadTx()).getTxHandle().getStoreTransaction()).getWrappedTransaction());
     }