HP SQLMX Publish Subscribe Functionality

torlee

What Is Publish/Subscribe?

Traditionally, relational database systems are passive data stores. Applications can
only retrieve and update data. Applications cannot request notification when new data
becomes available or when existing data is updated. However, immediate notification
of additions and updates to the database is essential for many applications.
Publish/subscribe is an event-driven mechanism in which a process known as a
subscriber receives notification of a published event from a process known as a
publisher. Queuing differs from publish/subscribe in that one, and only one, subscriber
consumes the published event or message. Queuing and publish/subscribe is often
used in applications in which a process needs to know of the existence of a timely
event, such the availability of a stock at a certain price or the arrival of a parcel at a
particular location.
Queuing and publish/subscribe services allow applications to receive notification of
changes to a database as soon as they occur. Applications might update and retrieve
data, but might need to be notified when data is updated or new data is added.
Publish/subscribe makes it possible to be notified in real time, and at less cost to the
system.Why Use Publish/Subscribe?


Suppose that you have a system to process new invoices. The invoices are stored in a
table, and a column called PROCESSED indicates unprocessed invoices. An
application processes these new invoices and updates the PROCESSED column to
show they have been processed. After all the invoices have been processed, the
application tries to fetch another unprocessed invoice and receives an end-of-file
condition.
Without publish/subscribe services, this process is more costly. To retrieve new
invoices that are inserted into the table after the end-of-file condition, the application
must close its cursor and reopen it, perhaps after waiting an interval, to see if new
invoices have arrived. This polling requires the closing and reopening of the cursor,
which has significant performance costs. It also requires application code to handle the
looping and end-of-file conditions.
You would probably code this application with an updatable cursor using a selection
predicate to include rows whose PROCESSED column is set to FALSE. You might also
use a positioned UPDATE statement to change the PROCESSED column to TRUE.
The application would need both of these statements to retrieve the information it
needs to process the invoice, and to make sure the invoice is processed only once. Asa result, every invoice would need two calls to the SQL executor, again at the expense
of extra performance costs and
application complexity.




If you simply started many instances of the application in an attempt to make this
application scalable, they would conflict with each other. As one instance held a lock
on the next invoice that needed processing, the others would have to wait. To avoid
conflict, you would have to devise a logical partitioning scheme so each instance
processed a subset of the invoices, adding complexity to the application and the
database schema.
Publish/Subscribe services provides solutions for each of these problems:
· Rather than polling for changes to the invoice table, the application can use stream
access mode. Stream access mode changes the behavior of the table: an attempt
to fetch a row blocks, if necessary, until more rows are available. The application
code becomes simpler and performs better.
· Instead of using a positioned UPDATE to mark the row as processed, theapplication can use a special type of statement, an embedded UPDATE, to select
and update a row in a single call. Again, the application code is simpler andperforms better.


· Publish/subscribe can use a special access mode, skip conflict access. As a result,
the various instances of the invoice processing application skip rows that are
locked and select and update other available rows. The application achieves better
performance without having to use a partitioning scheme.

Without Pub/Sub
A. retrieve record
B. process ... ...
C. update status

With Pub/Sub
A. retrieve record(Embeded update)
B. process ... ...

Publish/subscribe supports only audited tables.
Terminologys about Publish/Subscrib


Publishing, Inserting or updating rows of data in a queue or channel by using an
INSERT or UPDATE statement.
Subscribing, Subscribing to entries in a queue or channel by using a SELECT
statement specifying stream access to be notified when changes
occur.
Stream access, Changes the behavior of a table in a SELECT or embedded
UPDATE or DELETE statement. When no more rows satisfy the
selection predicate, the statement waits for more rows.
Queue, A database table read and updated in stream access mode, using an
UPDATE or DELETE statement embedded in the FROM clause of a
SELECT statement. Only a single subscriber will receive the new
entry.
Channel, A database table used by applications specifying stream access to
subscribe to newly published entries. The entries remain in the table.All subscribers receive new entries.


Queuing enqueueing,Inserting entries into a queue by using an INSERT or UPDATE
statement.
Dequeueing, Reading and deleting entries with a single operation by using a
SELECT statement with an embedded DELETE or UPDATE. This
dequeue operation is sometimes referred to as a destructive
SELECT. For another description of a destructive SELECT
Semi-queuing,Reading a queue by skipping rows that are locked by anothertransaction so that transactions do not block each other.


Major Features

· Stream access mode
· Embedded DELETE and UPDATE statements
· Skip conflict access mode
· Set on rollback
· Holdable cursors
· Rowset integration
· Horizontally partitioned tables· Ordered streams
1. Stream Access



    It allows an application to access SQL/MP or SQL/MX database tables as continuous data streams.
The stream access mode first causes a regular scan of the table and, after all available
rows have been returned, causes fetch operations to wait (block) instead of returning
the end-of-data condition. The fetch operation resumes when new rows becomeavailable.

• CREATE TABLE sos.ecui.quotes
  
• (symbol CHAR(5), price INT);
  
• 
  
• SET NAMETYPE ANSI;
  
• SET SCHEMA sos.ecui;
  
• SELECT * FROM STREAM(quotes);
  
• //This application is in a wait state until rows are inserted into the table.
  
• 
  
• SYMBOL  PRICE
  
• ------  -----------
  
• 
  
• IBM              10
  
• APP              11
  
• 
  
• >>INSERT INTO quotes VALUES('IBM',10);
  
• 
  
• ---1 row(s) inserted.
  
• >>INSERT INTO quotes VALUES('APP',11);
  
• 
  
• ---1 row(s) inserted.
  
• 
  
• UntilBreake it.
  
• 
  
• *** WARNING[15033]Break was received.
  
• 
  
• 
  
• *** WARNING[15018]Break was received.  Thelast statement may be lost.
  

Another way to stop it, set the timeout for the stream. the default value is -1.

The STREAM_TIMEOUT value is set to 300 in hundredths of seconds, which isequivalent to 3 seconds.  When the stream times out, control is returned to the application.
If the application is MXCI, the user will not be able to use the cursor, even though it is still opened. This is
because MXCI does not allow the user to control (open, fetch, and close) cursorsdirectly. However, a user-written application can use the cursor to fetch again.

• >>CONTROL QUERY DEFAULT stream_timeout '300';
  
• 
  
• --- SQL operation complete.
  
• >>SELECT * FROM STREAM(quotes);
  
• 
  
• SYMBOL  PRICE
  
• ------  -----------
  
• 
  
• IBM              10
  
• APP              11
  
• 
  
• *** ERROR[8006]The stream timed out, but the cursor is still open.
  
• 
  
• ---2 row(s) selected.
  

2. Embedded DELETE



Embedded DELETE statements allow applications to read and delete rows with a
single operation. Dequeue operations are implemented by using an embeddedDELETE together with stream access.
Destructive SELECTs，
Support for SQL/MP and SQL/MX Tables

• CREATE TABLE ecui.invoices
  
• (contractnbr INT, amount INT, priority INT);
  
• 
  
• INSERT INTO invoices VALUES(100,10500,1);
  
• INSERT INTO invoices VALUES(200,20390,2);
  
• INSERT INTO invoices VALUES(300,30800,3);
  
• 
  
• >>SELECT * FROM
  
• +>(DELETE FROM STREAM(invoices)) AS invoices;
  
• 
  
• CONTRACTNBR  AMOUNT       PRIORITY
  
• -----------  -----------  -----------
  
• 
  
•         100        10500            1
  
•         200        20390            2
  
•         300        30800            3
  
• 
  
• *** ERROR[8006]The stream timed out, but the cursor is still open.
  
• 
  
• ---3 row(s) selected.
  
• >>SELECT * FROM invoices;
  
• 
  
• ---0 row(s) selected.
  

The available rows are retrieved and deleted.The application is now in a wait stateuntil more rows are inserted into the table. or the stream timeout. And, we can use RollBack to cancle the Deletion. Because the test will delete the records, so we will control the transaction manurally. so we need to know something about BEGIN WORK statment;
　　The BEGIN WORK statement enables you to start a transaction explicitly—where the transaction consists of the set of operations defined by the sequence of SQL statements that begins immediately after BEGIN WORK and ends with the next COMMIT WORK or ROLLBACK WORK statement. The BEGIN WORK statement has no effect on nonaudited tables.
　　the structure is asbelow,
　　

• ---This statement initiates a transaction.
  
• BEGIN WORK;
  
• --- SQL operation complete.
  
• 
  
• there are some other SQL operations......
  
• ---This statement ends a transaction.
  
• COMMIT WORK;/ ROLLBACK WORK;
  
• --- SQL operation complete.
  
• 
  
• CONTROL QUERY DEFAULT stream_timeout '300';
  
• BEGIN WORK;
  
• SELECT * FROM
  
• (DELETE FROM STREAM(invoices)) AS invoices;
  
• CONTRACTNBR AMOUNT PRIORITY
  
• ---------------------------------
  
• 100105001
  
• 200203902
  
• 300308003
  
• *** ERROR[8006]The stream timed out,
  
• but the cursor is still open.
  
• ROLLBACK WORK;
  
• SELECT * FROM invoices;
  
• CONTRACTNBR AMOUNT PRIORITY
  
• ---------------------------------
  
• 100105001
  
• 200203902
  
• 300308003
  
• ---3 row(s) selected.
  
• 
  
• 
  
• the differences between "delete form table_name where ......."?
  
• Directlydelete just delete it, cannot get the value in this record and process it.
  
• With the Pub/Sub, we can get this record and then process it and delete it at last.
  

3. Embedded UPDATE

Embedded UPDATE statements enable applications to read and update rows with a
single operation.

• CREATE TABLE ecui.orders
  
• (order_nbr INT,
  
• amount INT,
  
• status CHAR(1) DEFAULT 'N');
  
• 
  
• INSERT INTO ORDERS(ORDER_NBR,AMOUNT) VALUES(1,500);
  
• INSERT INTO ORDERS(ORDER_NBR,AMOUNT) VALUES(2,4500);
  
• INSERT INTO ORDERS(ORDER_NBR,AMOUNT) VALUES(3,2100);
  
• 
  
• >>SELECT * FROM ORDERS;
  
• 
  
• ORDER_NBR    AMOUNT       STATUS
  
• -----------  -----------  ------
  
• 
  
•           1          500  N
  
•           2         4500  N
  
•           3         2100  N
  
• 
  
• ---3 row(s) selected.
  
• 
  
• >>SELECT * FROM
  
• +>(UPDATE STREAM(orders) SET status ='Y' WHERE status ='N')
  
• +>AS orders;
  
• 
  
• ORDER_NBR    AMOUNT       STATUS
  
• -----------  -----------  ------
  
• 
  
•           1          500  Y
  
•           2         4500  Y
  
•           3         2100  Y
  
• 
  
• *** ERROR[8006]The stream timed out, but the cursor is still open.
  
• 
  
• ---3 row(s) selected.
  

The differences between (update table_name set ..... where .....).4. Skip Conflict Access


The skip conflict access method is important in implementing an efficient transactional
queuing feature. It prevents concurrent transactions from blocking each other by
waiting for the release of locks on rows currently being inserted or updated while otherrows are available that are not currently locked by another transaction.

• INSERT INTO invoices VALUES(100,10500,1);
  
• INSERT INTO invoices VALUES(200,20390,2);
  
• INSERT INTO invoices VALUES(300,30800,3);
  
• 
  
• CONTROL QUERY DEFAULT stream_timeout '-1';
  
• SET NAMETYPE ANSI;
  
• SET SCHEMA sos.ecui;
  
• 
  
• >>SELECT * FROM
  
• +>(DELETE FROM STREAM(invoices)
  
• +>FOR SKIP CONFLICT ACCESS) AS invoices;
  
• 
  
• CONTRACTNBR  AMOUNT       PRIORITY
  
• -----------  -----------  -----------
  
• 
  
•         100        10500            1
  
•         200        20390            2
  
•         300        30800            3
  

in another session

• CONTROL QUERY DEFAULT stream_timeout '-1';
  
• SET SCHEMA sos.ecui;
  
• SELECT * FROM
  
• (DELETE FROM STREAM(invoices)
  
• FOR SKIP CONFLICT ACCESS) AS invoices;
  

in the 3rd session,

• CONTROL QUERY DEFAULT stream_timeout '-1';
  
• SET NAMETYPE ANSI;
  
• SET SCHEMA sos.ecui;
  
• INSERT INTO invoices VALUES(600,54800,6);
  

then we can find the 1st session get this new record

• CONTRACTNBR  AMOUNT       PRIORITY
  
• -----------  -----------  -----------
  
• 
  
•         100        10500            1
  
•         200        20390            2
  
•         300        30800            3
  
•         600        54800            6
  

5. AFTER LAST ROW

The stream skips all existing rows and returns rows published after the stream's cursorwas opened.

• >>SELECT * FROM INVOICES;
  
• 
  
• CONTRACTNBR  AMOUNT       PRIORITY
  
• -----------  -----------  -----------
  
• 
  
•         100        10500            1
  
•         200        20390            2
  
•         300        30800            3
  
• 
  
• ---3 row(s) selected.
  
• 
  
• CONTROL QUERY DEFAULT stream_timeout '-1';
  
• SELECT * FROM STREAM(invoices) AFTER LAST ROW;
  

In another session, we insert two records,

• >>INSERT INTO invoices VALUES(400,54800,4);
  
• 
  
• ---1 row(s) inserted.
  
• >>INSERT INTO invoices VALUES(400,54800,5);
  
• 
  
• ---1 row(s) inserted.
  

Meanwhile, in the 1st session, those two records will be retrieved.

Note that in the case where a publisher's transaction was in progress
when the subscriber’s statement began, not all the transaction's rows would be
returned to the subscriber. Instead only the rows inserted or updated after thestatement began executing would be returned.

• SELECT * FROM STREAM(invoices) AFTER LAST ROW;
  
• >>SELECT * FROM STREAM(invoices) AFTER LAST ROW;
  
• 
  
• CONTRACTNBR  AMOUNT       PRIORITY
  
• -----------  -----------  -----------
  
• 
  
•         400        54800            4
  
•         400        54800            5
  

6. Set Column Values On Rollback


Allows an application to update columns when
aborting a transaction during an embedded
DELETE or UPDATE operation. This field should be "NOT NULL"

• CREATE TABLE TEST_COL
  
• (
  
• ID INT DEFAULT NULL,
  
• NAME CHAR(20) DEFAULT NULL,
  
• COUNTER INT NOT NULL
  
• );
  
• 
  
• INSERT INTO TEST_COL VALUES(1,'ENDER',2);
  
• INSERT INTO TEST_COL VALUES(2,'ROCKY',2);
  
• 
  
• CONTROL QUERY DEFAULT stream_timeout '300';
  
• COMMIT WORK or ROLLBACK WORK
  
• 
  
• >>select * from TEST_COL;
  
• 
  
• ID           NAME                  COUNTER
  
• -----------  --------------------  -----------
  
• 
  
•           1  ENDER                           2
  
•           2  ROCKY                           2
  
• 
  
• ---2 row(s) selected.
  
• >>BEGIN WORK;
  
• 
  
• --- SQL operation complete.
  
• >>SELECT * FROM(DELETE FROM STREAM(TEST_COL)
  
• +>SET ON ROLLBACK COUNTER = COUNTER +1
  
• +>FOR SKIP CONFLICT ACCESS) AS TEST_COL;
  
• 
  
• ID           NAME                  COUNTER
  
• -----------  --------------------  -----------
  
• 
  
•           1  ENDER                           2
  
•           2  ROCKY                           2
  
• 
  
• *** ERROR[8006]The stream timed out, but the cursor is still open.
  
• 
  
• ---2 row(s) selected.
  
• >>select * from TEST_COL;
  
• 
  
• ---0 row(s) selected.
  
• >>ROLLBACK WORK;
  
• 
  
• --- SQL operation complete.
  
• >>select * from TEST_COL;
  
• 
  
• ID           NAME                  COUNTER
  
• -----------  --------------------  -----------
  
• 
  
•           1  ENDER                           3
  
•           2  ROCKY                           3
  
• 
  
• ---2 row(s) selected.
  

7. Holdable Cursors
A holdable cursor enables an application to retain an open cursor across transactions.

This feature is supported only for cursors using the stream access mode or for cursorsdefined with a table reference that uses an embedded UPDATE or DELETE.

The WITH HOLD clause can also be used with dynamic cursors, extended dynamiccursors, and allocated cursors


Below DECLARE CURSOR get_invoices,
statement specifies the cursor is to remain open and maintain its position in the resultset even though a user-specified transaction has terminated。

• EXEC SQL DECLARE get_invoices CURSOR WITH HOLD FOR
  
• SELECT amount,contractnbr
  
• FROM
  
• (DELETE FROM STREAM(invoices)
  
• FOR SKIP CONFLICT ACCESS) AS invoices;
  

Because the cursor was declared WITH HOLD, it is not closed by the execution of the
COMMIT WORK statement. Only an error condition or a concurrent DDL or utility
operation can cause a cursor to close. For example, an error such as a system or disk
process failure that would cause any cursor to close, causes a holdable cursor to
close. Before using a cursor closed by an error condition, the application must reopen
the cursor.

Applications that delete or update many rows, using embedded DELETEs and
UPDATEs, experience a significant performance boost if they use a holdable cursor
and are coded to avoid closing and reopening the cursor with each COMMIT WORKstatement.8. Rowset Integration


The integration of rowsets into the queuing and publish/subscribe features of SQL/MX
enables applications to enqueue or dequeue multiple rows at one time instead ofhaving to enqueue or dequeue rows one at a time.
Note, not supported in Java applications.
The cursor fetch into a rowset from a stream waits until the rowset is filled.


The rowsets are retrieved ten (or fewer than ten) at a time until the application mustwait for new arrivals.
9. Partitioned Queues
10. Ordered Streams
11. Joins
12. Run-Time Limits on Streams
13. Restarting a Subscriber
14. Embedded SQL


　　


来自为知笔记(Wiz)