PeriodicExtended { @Procedure(mode = Mode.WRITE) @Deprecated @Description("apoc.periodic.rock_n_roll_while('some cypher for knowing when to stop', 'some cypher for iteration', 'some cypher as action on each iteration', 10000) YIELD batches, total - run the action statement in batches over the iterator statement's results in a separate thread. Returns number of batches and total processed rows") public Stream<LoopingBatchAndTotalResult> rock_n_roll_while( @Name("cypherLoop") String cypherLoop, @Name("cypherIterate") String cypherIterate, @Name("cypherAction") String cypherAction, @Name("batchSize") long batchSize) { Map<String, String> fieldStatement = Util.map( "cypherLoop", cypherLoop, "cypherIterate", cypherIterate); validateQueries(fieldStatement); Stream<LoopingBatchAndTotalResult> allResults = Stream.empty(); Map<String,Object> loopParams = new HashMap<>(1); Object value = null; while (true) { loopParams.put("previous", value); try (Result result = tx.execute(cypherLoop, loopParams)) { value = result.next().get("loop"); if (!Util.toBoolean(value)) return allResults; } log.info("starting batched operation using iteration `%s` in separate thread", cypherIterate); try (Result result = tx.execute(cypherIterate)) { Stream<BatchAndTotalResult> oneResult = iterateAndExecuteBatchedInSeparateThread((int) batchSize, false, false,0, result, (tx, params) -> tx.execute(cypherAction, params), 50, -1); final Object loopParam = value; allResults = Stream.concat(allResults, oneResult.map(r -> r.inLoop(loopParam))); } } }  @Procedure(mode = Mode.WRITE) @Deprecated @Description("apoc.periodic.rock_n_roll_while('some cypher for knowing when to stop', 'some cypher for iteration', 'some cypher as action on each iteration', 10000) YIELD batches, total - run the action statement in batches over the iterator statement's results in a separate thread. Returns number of batches and total processed rows") Stream<LoopingBatchAndTotalResult> rock_n_roll_while(
            @Name("cypherLoop") String cypherLoop,
            @Name("cypherIterate") String cypherIterate,
            @Name("cypherAction") String cypherAction,
            @Name("batchSize") long batchSize); @Deprecated @Procedure(mode = Mode.WRITE) @Description("apoc.periodic.rock_n_roll('some cypher for iteration', 'some cypher as action on each iteration', 10000) YIELD batches, total - run the action statement in batches over the iterator statement's results in a separate thread. Returns number of batches and total processed rows") Stream<BatchAndTotalResult> rock_n_roll(
            @Name("cypherIterate") String cypherIterate,
            @Name("cypherAction") String cypherAction,
            @Name("batchSize") long batchSize); @Context public GraphDatabaseService db; @Context public TerminationGuard terminationGuard; @Context public Log log; @Context public Pools pools; @Context public Transaction tx; }
@Test public void testRock_n_roll_while() throws Exception { db.executeTransactionally("UNWIND range(1,100) AS x CREATE (:Person{name:'Person_'+x})"); testResult(db, "CALL apoc.periodic.rock_n_roll_while('return coalesce($previous,3)-1 as loop', 'match (p:Person) return p', 'MATCH (p) where p=$p SET p.lastname =p.name', 10)", result -> { long l = 0; while (result.hasNext()) { Map<String, Object> row = result.next(); assertEquals(2L - l, row.get("loop")); assertEquals(10L, row.get("batches")); assertEquals(100L, row.get("total")); l += 1; } assertEquals(2L, l); }); testCall(db, "MATCH (p:Person) where p.lastname is not null return count(p) as count", row -> assertEquals(100L, row.get("count")) ); }