Periodic { @Procedure(mode = Mode.WRITE) @Description("apoc.periodic.iterate('statement returning items', 'statement per item', {batchSize:1000,iterateList:true,parallel:false,params:{},concurrency:50,retries:0}) YIELD batches, total - run the second statement for each item returned by the first statement. Returns number of batches and total processed rows") public Stream<BatchAndTotalResult> iterate( @Name("cypherIterate") String cypherIterate, @Name("cypherAction") String cypherAction, @Name("config") Map<String,Object> config) { validateQuery(cypherIterate); long batchSize = Util.toLong(config.getOrDefault("batchSize", 10000)); int concurrency = Util.toInteger(config.getOrDefault("concurrency", 50)); boolean parallel = Util.toBoolean(config.getOrDefault("parallel", false)); BatchMode batchMode = BatchMode.fromConfig(config); long retries = Util.toLong(config.getOrDefault("retries", 0)); Map<String,Object> params = (Map<String, Object>) config.getOrDefault("params", Collections.emptyMap()); int failedParams = Util.toInteger(config.getOrDefault("failedParams", -1)); try (Result result = tx.execute(slottedRuntime(cypherIterate),params)) { Pair<String,Boolean> prepared = PeriodicUtils.prepareInnerStatement(cypherAction, batchMode, result.columns(), "_batch"); String innerStatement = prepared.first(); boolean iterateList = prepared.other(); log.info("starting batching from `%s` operation using iteration `%s` in separate thread", cypherIterate,cypherAction); return iterateAndExecuteBatchedInSeparateThread((int)batchSize, parallel, iterateList, retries, result, (tx, p) -> Iterators.count(tx.execute(innerStatement, merge(params, p))), concurrency, failedParams); } }  @Procedure @Description("apoc.periodic.list - list all jobs") Stream<JobInfo> list(); @Procedure(mode = Mode.WRITE) @Description("apoc.periodic.commit(statement,params) - runs the given statement in separate transactions until it returns 0") Stream<RundownResult> commit(@Name("statement") String statement, @Name(value = "params", defaultValue = "{}") Map<String,Object> parameters); @Procedure @Description("apoc.periodic.cancel(name) - cancel job with the given name") Stream<JobInfo> cancel(@Name("name") String name); @Procedure(mode = Mode.WRITE) @Description("apoc.periodic.submit('name',statement) - submit a one-off background statement") Stream<JobInfo> submit(@Name("name") String name, @Name("statement") String statement); @Procedure(mode = Mode.WRITE) @Description("apoc.periodic.repeat('name',statement,repeat-rate-in-seconds, config) submit a repeatedly-called background statement. Fourth parameter 'config' is optional and can contain 'params' entry for nested statement.") Stream<JobInfo> repeat(@Name("name") String name, @Name("statement") String statement, @Name("rate") long rate, @Name(value = "config", defaultValue = "{}") Map<String,Object> config ); @Procedure(mode = Mode.WRITE) @Description("apoc.periodic.countdown('name',statement,repeat-rate-in-seconds) submit a repeatedly-called background statement until it returns 0") Stream<JobInfo> countdown(@Name("name") String name, @Name("statement") String statement, @Name("rate") long rate); JobInfo submit(String name, Runnable task, Log log); JobInfo schedule(String name, Runnable task, long delay, long repeat); @Procedure(mode = Mode.WRITE) @Description("apoc.periodic.iterate('statement returning items', 'statement per item', {batchSize:1000,iterateList:true,parallel:false,params:{},concurrency:50,retries:0}) YIELD batches, total - run the second statement for each item returned by the first statement. Returns number of batches and total processed rows") Stream<BatchAndTotalResult> iterate(
            @Name("cypherIterate") String cypherIterate,
            @Name("cypherAction") String cypherAction,
            @Name("config") Map<String,Object> config); static final Pattern RUNTIME_PATTERN; static final Pattern CYPHER_PREFIX_PATTERN; static final String CYPHER_RUNTIME_SLOTTED; @Context public GraphDatabaseService db; @Context public TerminationGuard terminationGuard; @Context public Log log; @Context public Pools pools; @Context public Transaction tx; }
@Test public void testIterate() throws Exception { db.executeTransactionally("UNWIND range(1,100) AS x CREATE (:Person{name:'Person_'+x})"); testResult(db, "CALL apoc.periodic.iterate('match (p:Person) return p', 'SET p.lastname =p.name REMOVE p.name', {batchSize:10,parallel:true})", result -> { Map<String, Object> row = Iterators.single(result); assertEquals(10L, row.get("batches")); assertEquals(100L, row.get("total")); }); testCall(db, "MATCH (p:Person) where p.lastname is not null return count(p) as count", row -> assertEquals(100L, row.get("count")) ); }