Repository: neo4j-contrib/neo4j-apoc-procedures
Pull request #87 (2016-08-12 14:12:46)

#84 timeboxed execution of a cypher statement


Patch:
diff --git a/docs/cypher.adoc b/docs/cypher.adoc
index 438d396f3..3d8643091 100644
--- a/docs/cypher.adoc
+++ b/docs/cypher.adoc
@@ -39,3 +39,16 @@ call db.labels() yield label
 call apoc.cypher.run("match (:`"+label+"`) return count(*) as count", null) yield value
 return label, value.count as count
 ----
+
+== running a cypher statement timeboxed
+
+There's a way to terminate a cypher statement if it takes longer than a given threshold. Consider an expensive statement calculating cross product of shortestpaths for each pair of nodes:
+
+[source,cypher]
+----
+call apoc.cypher.runTimeboxed("match (n),(m) match p=shortestPath((n)-[*]-(m)) return p", null, 10000) yield value
+return value.p
+----
+
+
+
diff --git a/readme.adoc b/readme.adoc
index 670217d29..25e3db4e8 100644
--- a/readme.adoc
+++ b/readme.adoc
@@ -446,6 +446,7 @@ apoc.graph.from(data,'name',{properties}) yield graph | creates a virtual graph
 | CALL apoc.cypher.runMany('cypher;\nstatements;',{params}) | runs each semicolon separated statement and returns summary - currently no schema operations
 | CALL apoc.cypher.mapParallel(fragment, params, list-to-parallelize) yield value | executes fragment in parallel batches with the list segments being assigned to _
 | CALL apoc.cypher.doIt(fragment, params) yield value | executes writing fragment with the given parameters
+| CALL apoc.cypher.runTimeboxed('cypherStatement',{params}, timeout) | abort statement after timeout millis if not finished
 |===
 
 // end::cypher[]
diff --git a/src/main/java/apoc/cypher/Cypher.java b/src/main/java/apoc/cypher/Cypher.java
index c67d4c54f..792f3d3df 100644
--- a/src/main/java/apoc/cypher/Cypher.java
+++ b/src/main/java/apoc/cypher/Cypher.java
@@ -10,7 +10,9 @@ import org.neo4j.graphdb.Result;
 import org.neo4j.helpers.collection.Iterables;
 import org.neo4j.helpers.collection.Iterators;
 import org.neo4j.kernel.api.KernelTransaction;
+import org.neo4j.kernel.api.exceptions.Status;
 import org.neo4j.kernel.internal.GraphDatabaseAPI;
+import org.neo4j.logging.Log;
 import org.neo4j.procedure.Context;
 import org.neo4j.procedure.Name;
 import org.neo4j.procedure.PerformsWrites;
@@ -28,6 +30,7 @@ import java.util.stream.StreamSupport;
 import static apoc.util.MapUtil.map;
 import static java.lang.String.format;
 import static java.lang.String.join;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static java.util.stream.Collectors.toList;
 
 /**
@@ -46,6 +49,8 @@ public class Cypher {
     public GraphDatabaseAPI api;
     @Context
     public KernelTransaction tx;
+    @Context
+    public Log log;
 
     @Procedure
     @Description("apoc.cypher.run(fragment, params) yield value - executes reading fragment with the given parameters")
@@ -310,7 +315,7 @@ public class Cypher {
         public boolean tryAdvance(Consumer<? super T> action) {
             try {
                 if (tx.getReasonIfTerminated()!=null) return false;
-                T entry = queue.poll(100, TimeUnit.MILLISECONDS);
+                T entry = queue.poll(100, MILLISECONDS);
                 if (entry == null || entry == tombstone) return false;
                 action.accept(entry);
                 return true;
@@ -325,4 +330,18 @@ public class Cypher {
 
         public int characteristics() { return ORDERED | NONNULL; }
     }
+
+    @Procedure
+    @Description("apoc.cypher.runTimeboxed('cypherStatement',{params}, timeout) - abort statement after timeout ms if not finished")
+    public Stream<MapResult> runTimeboxed(@Name("cypher") String cypher, @Name("params") Map<String, Object> params, @Name("timeout") long timeout) {
+
+        Pools.SCHEDULED.schedule(() -> {
+            String txString = tx == null ? "<null>" : tx.toString();
+            log.warn("marking " + txString + " for termination");
+            tx.markForTermination(Status.Transaction.Terminated);
+        }, timeout, MILLISECONDS);
+
+        Result result = db.execute(cypher, params == null ? Collections.EMPTY_MAP : params);
+        return result.stream().map(stringObjectMap -> new MapResult(stringObjectMap));
+    }
 }


Test patch:
diff --git a/src/test/java/apoc/cypher/CypherTest.java b/src/test/java/apoc/cypher/CypherTest.java
index cf9b0d132..337a37ab7 100644
--- a/src/test/java/apoc/cypher/CypherTest.java
+++ b/src/test/java/apoc/cypher/CypherTest.java
@@ -1,17 +1,23 @@
 package apoc.cypher;
 
 import apoc.util.TestUtil;
+import apoc.util.Utils;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.ExpectedException;
 import org.neo4j.graphdb.GraphDatabaseService;
 import org.neo4j.graphdb.QueryExecutionException;
+import org.neo4j.graphdb.Result;
+import org.neo4j.graphdb.TransientTransactionFailureException;
 import org.neo4j.graphdb.factory.GraphDatabaseSettings;
 import org.neo4j.helpers.collection.Iterators;
 import org.neo4j.test.TestGraphDatabaseFactory;
 
 import java.io.File;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 
@@ -20,6 +26,7 @@ import static apoc.util.TestUtil.testCall;
 import static apoc.util.TestUtil.testResult;
 import static apoc.util.Util.toLong;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 
 /**
  * @author mh
@@ -29,6 +36,9 @@ public class CypherTest {
 
     private static GraphDatabaseService db;
 
+    @Rule
+    public ExpectedException thrown= ExpectedException.none();
+
     @BeforeClass
     public static void setUp() throws Exception {
         db = new TestGraphDatabaseFactory()
@@ -36,6 +46,7 @@ public class CypherTest {
                 .setConfig(GraphDatabaseSettings.load_csv_file_url_root,new File("src/test/resources").getAbsolutePath())
                 .newGraphDatabase();
         TestUtil.registerProcedure(db, Cypher.class);
+        TestUtil.registerProcedure(db, Utils.class);
     }
 
     @AfterClass
@@ -188,4 +199,12 @@ public class CypherTest {
                     assertEquals(false, r.hasNext());
                 });
     }
+
+    @Test(timeout=9000)
+    public void testWithTimeout() {
+        thrown.expect(TransientTransactionFailureException.class);
+        thrown.expectMessage("Explicitly terminated by the user.");
+        Result result = db.execute("CALL apoc.cypher.runTimeboxed('CALL apoc.util.sleep(10000)', null, {timeout})", Collections.singletonMap("timeout", 100));
+        assertFalse(result.hasNext());
+    }
 }
diff --git a/src/test/java/apoc/spatial/GeocodeTest.java b/src/test/java/apoc/spatial/GeocodeTest.java
index 27a9e2156..94bbc135a 100644
--- a/src/test/java/apoc/spatial/GeocodeTest.java
+++ b/src/test/java/apoc/spatial/GeocodeTest.java
@@ -101,9 +101,9 @@ public class GeocodeTest {
                 (row) -> {
                     Map value = (Map) row.get("location");
                     assertEquals("Incorrect latitude found", lat, Double.parseDouble(value.get("latitude").toString()),
-                            0.0005);
+                            0.1);
                     assertEquals("Incorrect longitude found", lon, Double.parseDouble(value.get("longitude").toString()),
-                            0.0005);
+                            0.1);
                 });
     }