Reactive SQL client and gRPC data streaming not working properly

[jdussouillez]

When testing some stuff related to another discussion (#36691), I realised that I can't get data streaming to work propertly between a SQL database and a gRPC client.

I have two apps to sync data:

• A gRPC server: read data from the db and send them to the client
• A gRPC client: read data from the gRPC response, group them by small chunks and insert them in another db

The problem is that I expected these two apps to streaming data from start to end :

• Server reads data in db using reactive client
• Send them in the gRPC response, while other data are read from the database (streaming)
• Client reads data from the gRPC response
• Client writes data in its database in small chunks

But the actual behavior is different: the server loads all the SQL results in the heap memory so with large data it ends with a OutOfMemory error.

Project to reproduce and error details: https://github.com/jdussouillez/quarkus-db-grpc-streaming

I'm not a Vert.x expert, but it seems that "row streaming" is supported: https://vertx.io/docs/vertx-pg-client/java/#_cursors_and_streaming
But I'm not using Vert.x directly (I'm using Quarkus + Mutiny)

[quarkus-bot[bot]]

/cc @alesj (grpc), @cescoffier (grpc), @tsegismont (reactive-sql-clients), @vietj (reactive-sql-clients)

[cescoffier]

This is the expected behavior. The getAll call in https://github.com/jdussouillez/quarkus-db-grpc-streaming/blob/master/server/server-rpc/src/main/java/com/github/jdussouillez/server/rpc/service/ProductApiService.java#L27 does not stream, because the database protocol does NOT stream (see https://foojay.io/today/optimizing-relational-database-access/).

If you need streaming, you need to use a cursor and understand the consequences (keeping a connection open for the lifetime of the cursor, sometimes even a transaction (it depends on the database))

[jdussouillez]

That's what I was afraid of... I was mislead by the use of Multi from start to end and thought the streaming was handled.

Thanks for the details!

[jdussouillez]

BTW, it would be nice to provide an example of streaming data using a cursor in the documentation.

Vert.x pg client requires a transaction for this case: https://vertx.io/docs/vertx-pg-client/java/#_cursors_and_streaming

PostreSQL destroys cursors at the end of a transaction, so the cursor API shall be used within a transaction, otherwise you will likely get the 34000 PostgreSQL error.

I managed to do something like this to test it and it works fine:

dbPool.getConnection()
    .call(connection -> connection.begin()
        .withContext((tr, ctx) -> tr.invoke(t -> ctx.put("t", t))) // TODO: find another way that this quick & dirty workaround to commit the transaction once all data are read from db
    )
    .chain(connection -> connection.prepare(getSQL(query)))
    .onItem()
    .transformToMulti(statement -> statement.createStream(500).toMulti())
    .map(mapper::apply)
    .withContext((multi, ctx) -> multi
        .onCompletion()
        .call(() -> ((Transaction) ctx.get("t")).commit())
    );

I also tried using Narayana's @Transactional annotation and also QuarkusTransaction.begin() but I can't because it blocks my thread: BlockingOperationNotAllowedException - Cannot start a JTA transaction from the IO thread.

As you said, using this implies that a connection from the pool and its transaction can stay open a long time (in our use case it's not a problem).