AQ support via R2DBC

[lukaseder]

Maybe it's premature, but I can hope :)

One awesome Oracle feature that could greatly profit from R2DBC support is Oracle AQ. Instead of blocking on DBMS_AQ.DEQUEUE calls, I could imagine a Publisher per queue that clients could subscribe to in various ways. Are there any plans for this yet, or too soon?

[Michael-A-McMahon]

Thanks for this suggestion! I haven't worked with AQ before, so my knowledge of this technology is really minimal. It certainly does seem that Publisher would be the most natural abstraction for an asynchronous queue, so that's something.

I'll plan to do some more research and just see what it would take to implement this. I'll also bring this to the attention of product management at Oracle and get their thoughts as well.

I'll update this issue once I know more.

[Michael-A-McMahon]

I don't feel that I've settled on a solution yet, but I'd like to share some thoughts I have at this point.

The R2DBC SPI is designed for SQL execution with a relational database. AQ requires APIs to enqueue and dequeue messages from a remote queue. These seem to represent two different use cases. So my first instinct is that while the R2DBC SPI is great for executing SQL, it might not be the greatest API for communicating with remote message queue.

One idea that I want to explore is what we can do with the javax.jms SPI. https://docs.oracle.com/javaee/7/api/javax/jms/MessageConsumer.html
This SPI seems to support asynchronous callbacks with setMessageListener(MessageListener). Asynchronous callbacks are one ingredient in the reactive streams recipe. The other ingredient is backpressure, and I'm not entirely sure if that is offered by the javax.jms SPI or not; I'm purely saying that I don't know the answer here, since I've never used javax.jms before. A final ingredient would be non-blocking network I/O, and I'm also not sure if we can assume a javax.jms implementation provides that or not.

Oracle AQ has a javax.jms implementation. If this implementation provides asynchronous callbacks, backpressure, and non-blocking I/O, then we should have all the ingredients need to implement a Publisher on top of this:
https://docs.oracle.com/en/database/oracle/oracle-database/19/jajms/index.html

At this point, I'm leaning towards a separate library designed for message queues as the right solution, rather than extending Oracle R2DBC. But I'd like to discuss more, and consider alternative ideas. As stated, I have no experience working with message queues, so I could benefit from hearing a perspective that is more informed about the message queue use case.

[lukaseder]

AQ requires APIs to enqueue and dequeue messages from a remote queue.

To the user, this is "just" (dreaded word, I know) a stored procedure call to some non-blocking version of DBMS_AQ.ENQUEUE and DBMS_AQ.DEQUEUE, which would allow for returning a Publisher<?> for enqueuing (returning some success codes, etc.) or Publisher<Result> similar to the execution of a query, returning some Row values. The ENQUEUE operation may even reside inside of some PL/SQL procedure, so it is transparent to the caller.

Perhaps this is no different from a procedure call to a procedure returning results via DBMS_SQL.RETURN_RESULT? The main difference being that unsubscribing from the AQ Publisher must affects the queue, whose rows must not be discarded, unlike those of DBMS_SQL.RETURN_RESULT. Or, if you will, the Publisher already pre-exists in the database, and clients may subscribe to it using DBMS_AQ.DEQUEUE calls.

In the past, I've found it challenging to integrate with AQ via JDBC, which was too bad because AQ is such useful technology. I really think this is a great use-case for both AQ and R2DBC, and a very distinct use-case from my past user point of view. I run a query (DBMS_AQ.DEQUEUE procedure call, of some sort (the current procedure blocks, so it won't work)), and I expect asynchronous results eventually. It almost works like a Kafka SQL style stream table. The use-cases are so numerous. Think of audit logging from a trigger that emits delta messages to some queue. These messages feel like an ordinary reactive SELECT message FROM messages WHERE .... I mean, thinking out loud, it would even be possible to add such features to the Oracle database itself. Create new SQL syntax that translates to DBMS_AQ.DEQUEUE calls behind the scenes and make this queuing transparent to users.

I don't think a JMS client is a bad idea. It fits well in an Java EE paradigm, but one approach doesn't exclude the other. I can't stress this enough: AQ is a very underrated piece of Oracle technology. PostgreSQL has this NOTIFY mechanism which is a (much) simplified version of AQ, and can be made to work with R2DBC (perhaps @mp911de can explain? https://github.com/mp911de/r2dbc-postgres-notification-example). Getting a similar example to work easily with Oracle AQ would add lots of value to Oracle customers, and help with the adoption of this underrated technology. I'm just thinking out loud, but I've always seen AQ as this Kafka SQL style stream. Not as flexible, but very similar, and with a lot of untapped potential.

[mp911de]

Re the Postgres example:
Postgres allows registering a LISTEN subscription for a connection. Once the server receives a Pub/Sub message via NOTIFY the server sends a message frame to each client connection that has an active subscription, the mechanism is basically backed by a server-side managed routing table.

All this makes only sense by the fact that the R2DBC Postgres driver uses a push-based I/O mechanism. The driver keeps reading incoming frames. Since notifications are not a response to a SQL command, these are generally made available through a Publisher<Notification>. Note that there's no ongoing client-side polling by issuing regular commands or so.

[lukaseder]

it would even be possible to add such features to the Oracle database itself

This is going to be out of scope for R2DBC, but I don't know where else to place this idea, except to send an email to @gvenzl :-)

Imagine this. Some new synthetic syntax like this:

STREAM a, b
FROM my_aq
WHERE x = 1

The actual syntax is not important. MY_AQ is a user-defined Oracle AQ created with DBMS_AQADM. There isn't much of a difference between MY_AQ and a reactive-streams Publisher.

This STREAM statement (instead of SELECT, but again, syntax is not important) would now produce a non-blocking DBMS_AQ.DEQUEUE subscription, which can be subscribed to by clients via a R2DBC Publisher, just like any other SQL statement. If necessary, the Oracle Database would create (on the server side) auxiliary queues that translate between MY_AQ and the synthetic, ad-hoc queue, e.g. to apply some filtering or projection that isn't currently possible via DBMS_AQ directly. There would obviously be limitations to what's possible, syntax wise, just like for materialized views.

And BAM. Oracle Database has its own Kafka SQL language, which would be a no-brainer to integrate with from R2DBC, all backed by AQ, but people wouldn't have to worry about the low level AQ details and laborious infrastructure logic anymore.

[mp911de]

FWIW, this is something a lot of folks keep asking, to obtain a live view of data (mostly speaking about new data being inserted) where a SELECT-like projection doesn't complete but the result cursor remains open (similar to MongoDB's tailable cursors).

[Douglas-Surber]

Oracle calls this Database Change Notification. Our reactive extensions to JDBC don't currently cover Database Change Notification, but they could be extended to do so. We could provide native support for an R2DBC API that provides this capability. Nothing here should be construed as making any commitment by Oracle. Douglas On Apr 27, 2021, at 12:12 AM, Mark Paluch ***@***.******@***.***>> wrote: FWIW, this is something a lot of folks keep asking, to obtain a live view of data (mostly speaking about new data being inserted) where a SELECT-like projection doesn't complete but the result cursor remains open (similar to MongoDB's tailable cursors). — You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub<https://urldefense.com/v3/__https://github.com/oracle/oracle-r2dbc/issues/27*issuecomment-827371993__;Iw!!GqivPVa7Brio!MyNbwg2WaJ2DadDDiS3pToBulZSam_55WTJKUrx9fSBZPLpJOA1E68pZle7sgzG7Bt4$>, or unsubscribe<https://urldefense.com/v3/__https://github.com/notifications/unsubscribe-auth/AARZS6OLNFDLTA7ZSWQRTK3TKZPXPANCNFSM43BBSJ6A__;!!GqivPVa7Brio!MyNbwg2WaJ2DadDDiS3pToBulZSam_55WTJKUrx9fSBZPLpJOA1E68pZle7s-AL6N4g$>.