CypherExtended { @Procedure @Description("apoc.cypher.mapParallel(fragment, params, list-to-parallelize) yield value - executes fragment in parallel batches with the list segments being assigned to _") public Stream<MapResult> mapParallel(@Name("fragment") String fragment, @Name("params") Map<String, Object> params, @Name("list") List<Object> data) { final String statement = withParamsAndIterator(fragment, params.keySet(), "_"); tx.execute("EXPLAIN " + statement).close(); return Util.partitionSubList(data, PARTITIONS,null) .flatMap((partition) -> Iterators.addToCollection(tx.execute(statement, parallelParams(params, "_", partition)), new ArrayList<>(partition.size())).stream()) .map(MapResult::new); }  @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 testMapParallel() throws Exception { int size = 10_000; testResult(db, "CALL apoc.cypher.mapParallel('UNWIND range(0,9) as b RETURN b',{},range(1,$size))", map("size", size), r -> assertEquals( size * 10,Iterators.count(r) )); }