Run multiple reads in parallel

Drivers & Stacks Python
1

Hi, I'm using neo4j community and one of my usecases is to expose a search endpoint which accepts a list of queries which then are executed against neo4j. The number of queries varies but is usually in the 50-100 range sometimes a bit more but we can control this.

They all use the same cypher query but the parameter differs as I send each individual term.

Currently I do the following:

await asyncio.gather(*[search(t) for t in terms])

Where search has the following

await self.driver.execute_query(query, database_=self.database, term=term)
  • Driver is reused
  • Connected using bolt

For this particular usecase the performance of the code really degrades inside the asyncio.gather call. Running the same code but via the http api gives better results.

What is the best way to approach this problem. Can we execute multiple reads inside a single transaction? Would it make sense to chunk an incoming request with 50 items into 5 sessions doing 10 reads each?

Running a single search via http takes 8ms, bolt is at 21.

2

I don't know what you query looks like, but let's assume something like this:

async def search(drvier, term):
    records, _, _ = await driver.execute_query(
        """
        MATCH (n: {term: $term})
        RETURN collect(n.result) AS results
        """,
        database_="neo4j",
        term=term,
    )
    return records[0]["results"]

Instead, you could roll the loop into cypher, saving you a lot of round trips

async def search(drvier, terms):
    records, _, _ = await driver.execute_query(
        """
        UNWIND $terms AS term
        MATCH (n:Thing {term: term})
        WITH term, collect(n.result)  AS results
        RETURN results
        """,
        database_="neo4j",
        terms=terms,
    )
    return [r["results"] for r in records]

If that isn't an option for you, you can indeed run multiple queries per transaction. For that, you'll have to use the session API of the driver.

async def work(tx):
    await tx.run("query1")
    await tx.run("query2")

async with driver.session(database="neo4j") as session:
    await session.execute_read(work)

Please see the API docs for details on that.