Rowtime Bounds API
The issues of handling rowtime bounds arise in several contexts of interfacing streaming SQL applications.
Rowtime bounds cannot be read or written in SQL.
However, rowtime bounds can be read or written in Java clients using the JDBC Driver as well as in adapters using the SQLstream API defined in the SDK, as described below.
More detailed descriptions appear in the Integration Guide JDBC Driver topic about setting and getting rowtime bounds. The relevant functions interfaces are mentioned in the section titled setting and getting rowtime bounds in the context of merging and rolling/sliding windows; their interfaces are shown in that book's section titled Extensions to JDBC API. and an example appears in the section titled Sending Rowtime Bounds.
Many applications use pumps as a way of inserting the results of a query into a stream. There is no way in SQL for a pump to punctuate, that is, to create rowtime bounds. A pump can only insert rows. However, the pump will pass forward the rowtime bounds it finds in its data source. (This point is explained in the Eliminating Rows topic of this guide.)
Java client applications connect to the s-Server using the SQLstream JDBC Driver, which has been extended to support streaming concepts. In particular, java.sql.Statement is extended as StreamingStatement, and java.sql.PreparedStatement as StreamingPreparedStatement.
Typically a data source will prepare an INSERT statement with parameters, and execute the prepared statement several times, binding different parameter values. A data reader executes a SELECT STREAM statement. In JDBC this corresponds to a StreamingStatement. To read rows, the reader gets a ResultSet and calls ResultSet.next() in a loop. The ResultSet from a SELECT STREAM is essentially a result stream: each call of next() returns the next row in the stream, and may block, waiting for the next row to arrive.
The SQLstream JDBC driver lets a writer set the rowtime bound (lets it "punctuate" its stream of input rows), and lets a reader get the current rowtime bound (get the current stream clock for the stream of result rows for his query).
This is the interface:
// returns the current bound for the results from the statement
// Sets the rowtime bound. Must be an input statement.
// An error to set the bound backwards in time.
// long form: 'cal' indicates the timezone of 't'.
// Next row's rowtime must >= t.
// 'strict' makes the constraint strict: rowtime > t.
void StreamingPreparedStatement.setRowtimeBound(Timestamp t, Calendar cal, boolean strict);
// short form: non-strict bound, UTC timezone
void StreamingPreparedStatement.setRowtimeBound(Timestamp t);
|•||If possible, a data source inserting rows should call|
before a gap, to indicate the rowtime of the next row to be inserted.
|•||A data reader can get the stream clock by calling|
It returns a non-strict lower bound on future rows not yet in the JDBC client buffer.
An application may call getRowtimeBound() several times; the value returned may increase, which means that the system has provided the client with a stronger bound.
A UDF (User-Defined Function), coded in Java, can be used as an element of a scalar SQL expression. Thus it is used as part of a row-by-row transformation or a filter. In s-Server, a UDF passes on rowtime bounds unchanged.
When writing a UDX (User-Defined Transformation), you need to be aware of rowtime bounds. See the topic Rowtime Bounds and UDXs in the Integration Guide for more details.
An adapter is a Java module that is either a data source or a data sink. A data source calls the Java interface com.sqlstream.plugin.RowSink to send rows, and a data sink implements the same interface to receive rows.
This interface has a method to transfer a rowtime bound:
A data source can call this, to the same effect as a JDBC client data source calling StreamingPreparedStatement.setRowtimeBound A data sink may use rowtime bounds sent to it as hints about future inputs, which may allow it to do some computations in advance, in a way analogous to the merge operator discussed above.
The following lines of Java code provide an example of a producer introducing a bound with a later rowtime than the last occurring row. These lines produce the three orders shown in the Implicit Rowtime Bounds section, followed by an arbitrary later bound:
1. Connection connection01;
2. StreamingPreparedStatement pstmt = connection.prepareStatement(
3. "INSERT INTO Orders (rowtime, ticker) VALUES (?, ?)");
4. // Create a calendar for '2006-09-11 12:00:00'.
5. Calendar calendar01 = Calendar.getInstance(TimeZone.getTimeZone("GMT"));
6. calendar01.set(Calendar.YEAR, 2006);
7. calendar01.set(Calendar.MONTH, 09); // months are 0-based
8. calendar01.set(Calendar.DAY_OF_MONTH, 11);
9. calendar01.set(Calendar.HOUR_OF_DAY, 12);
10. calendar01.set(Calendar.MINUTE, 0);
11. calendar01.set(Calendar.SECOND, 0);
12. // Send ORDERS: 'ORCL', rowtime '2006-09-11 12:00:00'.
13. pstmt.set(1, new Timestamp(calendar01.getTimeInMillis()));
14. pstmt.set(2, "ORCL");
16. // Send ORDERS: 'MSFT', rowtime '2006-09-11 12:01:00'.
17. calendar01.set(Calendar.MINUTE, 1);
18. pstmt.set(1, new Timestamp(calendar01.getTimeInMillis()));
19. pstmt.set(2, "MSFT");
21. // Send ORDERS: 'IBM', rowtime '2006-09-11 12:02:00'.
22. calendar01.set(Calendar.MINUTE, 2);
23. pstmt.set(1, new Timestamp(calendar01.getTimeInMillis()));
24. pstmt.set(2, "IBM");
26. // Send rowtime bound '2006-09-11 12:05:00'.
27. calendar01.set(Calendar.MINUTE, 5);
28. pstmt.setRowtimeBound(new Timestamp(calendar01.getTimeInMillis()));
The results of this code are as follows:
|•||Set up the calendar for the 10th month, 11th day, 12th hour (lines 4 through 11).|
|•||Create the first order, for ORCL, occurring exactly at 12:00:00 (lines 12 through 15).|
|•||Create the second order, for MSFT, occurring exactly at 12:01:00 (lines 16 - 20).|
|•||Create the third order, for IBM, occurring exactly at 12:02:00 (lines 21 - 25).|
|•||Create the producer's chosen arbitrary bound [2006-09-11 12:05:00] (lines 26 - 28).|
These 28 lines will create the previous sequence of rows, and follow it with the bound: ORDERS.bound: rowtime '2006-09-11 12:05:00'. The bound is a promise by the caller not to insert another row using this prepared statement with a rowtime earlier than 2006-09-11 12:05:00.