CypherExtended { @Procedure public Stream<MapResult> parallel2(@Name("fragment") String fragment, @Name("params") Map<String, Object> params, @Name("parallelizeOn") String key) { if (params == null) return Cypher.runCypherQuery(tx, fragment, params); if (key == null || !params.containsKey(key)) throw new RuntimeException("Can't parallelize on key " + key + " available keys " + params.keySet()); Object value = params.get(key); if (!(value instanceof Collection)) throw new RuntimeException("Can't parallelize a non collection " + key + " : " + value); final String statement = withParamsAndIterator(fragment, params.keySet(), key); tx.execute("EXPLAIN " + statement).close(); Collection<Object> coll = (Collection<Object>) value; int total = coll.size(); int partitions = PARTITIONS; int batchSize = Math.max(total / partitions, 1); if (batchSize > MAX_BATCH) { batchSize = MAX_BATCH; partitions = (total / batchSize) + 1; } List<Future<List<Map<String, Object>>>> futures = new ArrayList<>(partitions); List<Object> partition = new ArrayList<>(batchSize); for (Object o : coll) { partition.add(o); if (partition.size() == batchSize) { futures.add(submit(db, statement, params, key, partition, terminationGuard)); partition = new ArrayList<>(batchSize); } } if (!partition.isEmpty()) { futures.add(submit(db, statement, params, key, partition, terminationGuard)); } return futures.stream().flatMap(f -> { try { return f.get().stream().map(MapResult::new); } catch (InterruptedException | ExecutionException e) { throw new RuntimeException("Error executing in parallel " + statement, e); } }); }  @Procedure(mode = WRITE) @Description("apoc.cypher.runFile(file or url,[{statistics:true,timeout:10,parameters:{}}]) - runs each statement in the file, all semicolon separated - currently no schema operations") Stream<RowResult> runFile(@Name("file") String fileName, @Name(value = "config",defaultValue = "{}") Map<String,Object> config); @Procedure(mode = WRITE) @Description("apoc.cypher.runFiles([files or urls],[{statistics:true,timeout:10,parameters:{}}])) - runs each statement in the files, all semicolon separated") Stream<RowResult> runFiles(@Name("file") List<String> fileNames, @Name(value = "config",defaultValue = "{}") Map<String,Object> config); @Procedure(mode=Mode.SCHEMA) @Description("apoc.cypher.runSchemaFile(file or url,[{statistics:true,timeout:10}]) - allows only schema operations, runs each schema statement in the file, all semicolon separated") Stream<RowResult> runSchemaFile(@Name("file") String fileName, @Name(value = "config",defaultValue = "{}") Map<String,Object> config); @Procedure(mode=Mode.SCHEMA) @Description("apoc.cypher.runSchemaFiles([files or urls],{statistics:true,timeout:10}) - allows only schema operations, runs each schema statement in the files, all semicolon separated") Stream<RowResult> runSchemaFiles(@Name("file") List<String> fileNames, @Name(value = "config",defaultValue = "{}") Map<String,Object> config); static String withParamMapping(String fragment, Collection<String> keys); static String compiled(String fragment); @Procedure Stream<MapResult> parallel(@Name("fragment") String fragment, @Name("params") Map<String, Object> params, @Name("parallelizeOn") String key); @Procedure @Description("apoc.cypher.mapParallel(fragment, params, list-to-parallelize) yield value - executes fragment in parallel batches with the list segments being assigned to _") Stream<MapResult> mapParallel(@Name("fragment") String fragment, @Name("params") Map<String, Object> params, @Name("list") List<Object> data); @Procedure @Description("apoc.cypher.mapParallel2(fragment, params, list-to-parallelize) yield value - executes fragment in parallel batches with the list segments being assigned to _") Stream<MapResult> mapParallel2(@Name("fragment") String fragment, @Name("params") Map<String, Object> params, @Name("list") List<Object> data, @Name("partitions") long partitions,@Name(value = "timeout",defaultValue = "10") long timeout); Map<String, Object> parallelParams(@Name("params") Map<String, Object> params, String key, List<Object> partition); @Procedure Stream<MapResult> parallel2(@Name("fragment") String fragment, @Name("params") Map<String, Object> params, @Name("parallelizeOn") String key); static String withParamsAndIterator(String fragment, Collection<String> params, String iterator); static final String COMPILED_PREFIX; static final int PARTITIONS; static final int MAX_BATCH; @Context
    public Transaction tx; @Context
    public GraphDatabaseService db; @Context
    public Log log; @Context
    public TerminationGuard terminationGuard; @Context
    public Pools pools; }
@Test public void testParallel2() throws Exception { int size = 10_0000; List<Long> list = new ArrayList<>(size); for (int i = 0; i < size; i++) list.add(3L); testCall(db, "CALL apoc.cypher.parallel2('RETURN a + 7 as b',{a:$list},'a') YIELD value RETURN sum(value.b) as b", map("list", list), r -> { assertEquals( size * 10L, r.get("b") ); }); }