Merge pull request JanusGraph#78 from ngageoint/tinkerpop-upgrade

Update to TinkerPop 3.2.3

janusgraph-berkeleyje/src/test/java/org/janusgraph/blueprints/BerkeleyGraphComputerProvider.java

@@ -33,7 +33,8 @@ public class BerkeleyGraphComputerProvider extends AbstractJanusGraphComputerPro
 
     @Override
     public ModifiableConfiguration getJanusGraphConfiguration(String graphName, Class<?> test, String testMethodName) {
-        ModifiableConfiguration config = BerkeleyStorageSetup.getBerkeleyJEConfiguration(StorageSetup.getHomeDir(graphName));
+        ModifiableConfiguration config = super.getJanusGraphConfiguration(graphName, test, testMethodName);
+        config.setAll(BerkeleyStorageSetup.getBerkeleyJEConfiguration(StorageSetup.getHomeDir(graphName)).getAll());
         config.set(GraphDatabaseConfiguration.IDAUTHORITY_WAIT, Duration.ofMillis(20));
         config.set(GraphDatabaseConfiguration.STORAGE_TRANSACTIONAL,false);
         return config;

janusgraph-cassandra/src/test/java/org/janusgraph/blueprints/thrift/ThriftGraphComputerProvider.java

@@ -16,7 +16,6 @@
 
 import org.janusgraph.CassandraStorageSetup;
 import org.janusgraph.blueprints.AbstractJanusGraphComputerProvider;
-import org.janusgraph.blueprints.AbstractJanusGraphProvider;
 import org.janusgraph.diskstorage.configuration.ModifiableConfiguration;
 import org.janusgraph.graphdb.olap.computer.FulgoraGraphComputer;
 import org.apache.tinkerpop.gremlin.GraphProvider;
@@ -30,7 +29,9 @@ public class ThriftGraphComputerProvider extends AbstractJanusGraphComputerProvi
     @Override
     public ModifiableConfiguration getJanusGraphConfiguration(String graphName, Class<?> test, String testMethodName) {
         CassandraStorageSetup.startCleanEmbedded();
-        return CassandraStorageSetup.getCassandraThriftConfiguration(graphName);
+        ModifiableConfiguration config = super.getJanusGraphConfiguration(graphName, test, testMethodName);
+        config.setAll(CassandraStorageSetup.getCassandraThriftConfiguration(graphName).getAll());
+        return config;
     }
 
 }

janusgraph-core/src/main/java/org/janusgraph/core/JanusGraph.java

@@ -52,6 +52,16 @@
         test = "org.apache.tinkerpop.gremlin.process.computer.GraphComputerTest",
         method = "shouldProcessResultGraphNewWithPersistVertexProperties",
         reason = "The result graph should return an empty iterator when vertex.edges() or vertex.vertices() is called.")
+@Graph.OptOut(
+        test = "org.apache.tinkerpop.gremlin.structure.io.IoTest$GraphMLTest",
+        method = "shouldReadGraphMLWithNoEdgeLabels",
+        reason = "JanusGraph does not support default edge label (edge) used when GraphML is missing edge labels.")
+@Graph.OptOut(
+        test = "org.apache.tinkerpop.gremlin.process.computer.GraphComputerTest",
+        method = "shouldSupportGraphFilter",
+        reason = "JanusGraph test graph computer (FulgoraGraphComputer) " +
+            "currently does not support graph filters but does not throw proper exception because doing so breaks numerous " +
+            "tests in gremlin-test ProcessComputerSuite.")
 public interface JanusGraph extends Transaction {
 
    /* ---------------------------------------------------------------

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

@@ -26,6 +26,7 @@
 import org.janusgraph.diskstorage.keycolumnvalue.cache.KCVSCache;
 import org.janusgraph.diskstorage.log.kcvs.ExternalCachePersistor;
 import org.apache.commons.lang.StringUtils;
+import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalInterruptedException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -439,8 +440,14 @@ public String toString() {
 
 
     private final <V> V executeRead(Callable<V> exe) throws JanusGraphException {
-        return BackendOperation.execute(exe, maxReadTime);
+        try {
+            return BackendOperation.execute(exe, maxReadTime);
+        } catch (JanusGraphException e) {
+            // support traversal interruption
+            // TODO: Refactor to allow direct propagation of underlying interrupt exception
+            if (Thread.interrupted()) throw new TraversalInterruptedException();
+            throw e;
+        }
     }
 
-
 }

janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/scan/ScanJob.java

@@ -33,7 +33,7 @@ public interface ScanJob extends Cloneable {
     /**
      * Invoked before a block of computation (i.e. multiple process() calls) is handed to this particular ScanJob.
      * Can be used to initialize the iteration. This method is called exactly once for each before a block of computation.
-     * This method is semantically aligned with {@link com.tinkerpop.gremlin.process.computer.VertexProgram#workerIterationStart()}
+     * This method is semantically aligned with {@link org.tinkerpop.gremlin.process.computer.VertexProgram#workerIterationStart()}
      *
      * This method may not be called if there is no data to be processed. Correspondingly, the end method won't be called either.
      *
@@ -49,7 +49,7 @@ public default void workerIterationStart(Configuration jobConfiguration,
     /**
      * Invoked after a block of computation (i.e. multiple process() calls) is handed to this particular ScanJob.
      * Can be used to close any resources held by this job. This method is called exactly once for each after a block of computation.
-     * This method is semantically aligned with {@link com.tinkerpop.gremlin.process.computer.VertexProgram#workerIterationEnd()}
+     * This method is semantically aligned with {@link org.tinkerpop.gremlin.process.computer.VertexProgram#workerIterationEnd()}
      *
      * This method may not be called if there is no data to be processed. Correspondingly, the start method won't be called either.
      *

janusgraph-core/src/main/java/org/janusgraph/diskstorage/util/BackendOperation.java

@@ -90,6 +90,8 @@ public static final<V> V executeDirect(Callable<V> exe, Duration totalWaitTime)
                 try {
                     Thread.sleep(waitTime.toMillis());
                 } catch (InterruptedException r) {
+                    // added thread interrupt signal to support traversal interruption
+                    Thread.currentThread().interrupt();
                     throw new PermanentBackendException("Interrupted while waiting to retry failed backend operation", r);
                 }
             } else {

janusgraph-core/src/main/java/org/janusgraph/graphdb/database/serialize/StandardSerializer.java

@@ -41,6 +41,7 @@
 import org.janusgraph.graphdb.types.TypeDefinitionCategory;
 import org.janusgraph.graphdb.types.TypeDefinitionDescription;
 
+import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet;
 import org.apache.tinkerpop.gremlin.structure.Direction;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -133,6 +134,8 @@ public StandardSerializer() {
         registerClassInternal(64,Duration.class, new DurationSerializer());
         registerClassInternal(65,Instant.class, new InstantSerializer());
         registerClassInternal(66,StandardTransactionId.class, new StandardTransactionIdSerializer());
+        registerClassInternal(67,TraverserSet.class, new SerializableSerializer());
+        registerClassInternal(68,HashMap.class, new SerializableSerializer());
 
     }
 

janusgraph-core/src/main/java/org/janusgraph/graphdb/database/serialize/attribute/SerializableSerializer.java

@@ -0,0 +1,52 @@
+// Copyright 2017 JanusGraph Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package org.janusgraph.graphdb.database.serialize.attribute;
+
+import org.janusgraph.core.attribute.AttributeSerializer;
+import org.janusgraph.diskstorage.ScanBuffer;
+import org.janusgraph.diskstorage.WriteBuffer;
+import org.janusgraph.graphdb.database.serialize.DataOutput;
+import org.janusgraph.graphdb.database.serialize.Serializer;
+import org.janusgraph.graphdb.database.serialize.SerializerInjected;
+import org.apache.commons.lang3.SerializationUtils;
+
+import java.io.Serializable;
+
+/**
+ * Serializes {@link Serializable} objects.
+ * @param <T> Serializable type
+ */
+public class SerializableSerializer<T extends Serializable> implements AttributeSerializer<T>, SerializerInjected {
+
+    private Serializer serializer;
+
+    @Override
+    public T read(ScanBuffer buffer) {
+        byte[] data = serializer.readObjectNotNull(buffer,byte[].class);
+        return (T) SerializationUtils.deserialize(data);
+    }
+
+    @Override
+    public void write(WriteBuffer buffer, T attribute) {
+        DataOutput out = (DataOutput) buffer;
+        out.writeObjectNotNull(SerializationUtils.serialize(attribute));
+    }
+
+    @Override
+    public void setSerializer(Serializer serializer) {
+        this.serializer = serializer;
+    }
+
+}

janusgraph-core/src/main/java/org/janusgraph/graphdb/olap/VertexJobConverter.java

@@ -72,9 +72,10 @@ protected VertexJobConverter(JanusGraph graph, VertexScanJob job) {
     }
 
     protected VertexJobConverter(VertexJobConverter copy) {
-        this.graph = new GraphProvider();
-        if (copy.graph.isProvided()) this.graph.setGraph(copy.graph.get());
+        this.graph = copy.graph;
         this.job = copy.job.clone();
+        this.tx = copy.tx;
+        this.idManager = copy.idManager;
     }
 
     public static ScanJob convert(JanusGraph graph, VertexScanJob vertexJob) {
@@ -96,18 +97,22 @@ public static StandardJanusGraphTx startTransaction(StandardJanusGraph graph) {
 
     @Override
     public void workerIterationStart(Configuration jobConfig, Configuration graphConfig, ScanMetrics metrics) {
-        graph.initializeGraph(graphConfig);
-        idManager = graph.get().getIDManager();
         try {
-            tx = startTransaction(graph.get());
+            open(graphConfig);
             job.workerIterationStart(graph.get(), jobConfig, metrics);
         } catch (Throwable e) {
             close();
             throw e;
         }
     }
 
-    private void close() {
+    protected void open(Configuration graphConfig) {
+        graph.initializeGraph(graphConfig);
+        idManager = graph.get().getIDManager();
+        tx = startTransaction(graph.get());
+    }
+
+    protected void close() {
         if (null != tx && tx.isOpen())
             tx.rollback();
         graph.close();