6 Oracle CQL Processors

A stream channel inserts events into a collection and sends the stream to the next EPN stage. Events in a stream flow continuously, can never be deleted from the stream, and have no end. You can perform queries on the continuous stream of events flowing into your application.

A query on the input stream channel, StockTradeIChannel, to retrieve all stock trade events with the Oracle ticker symbol follows.

SELECT tickerSymbol
FROM StockTradeIStreamChannel
WHERE tickerSymbol = ORCL

The following configuration file entry shows the query. ISTREAM is a relation to stream operator described in Relation to Stream Operators.

<processor>
  <rules>
    <query id=rule1 <![CDATA[ISTREAM (SELECT tickerSymbol
       FROM StockTradeIStreamChannel WHERE tickerSymbol = ORCL)>
    </query>
  </rules>
</processor>

A relation channel inserts events into a collection and sends the relation to the next EPN stage. A relation is a window of time on the stream that has a beginning and an end. Events in a relation can be inserted into, deleted from, and updated in the relation. For insert, delete, and update operations, events in a relation must be referenced to a particular point in time to ensure the operation takes place on the correct event. All operations on a relation are time based.

Most applications do not use relation channels. You can put a window of time on events coming from a stream channel to create a relation for time-based processing operations. To find the average price for a particular stock, you must determine a time frame (window) in which to calculate the average. When you define a window on a stream, you have a collection of data that is not flowing, and unlike a stream, has a beginning and an end. The window is an in-memory relation on which you can apply a function such as AVG and also perform insert, update, and delete operations.

Operators that put a window of time on a stream are called stream to relation operators. The output of stream to relation operations are relations. You use relation to stream operators to convert a relation back to a stream to output a stream that contains every event, only inserted events, or only deleted events.

Oracle CQL processor output typically goes to a stream channel and on to the next stage in the EPN.

Instead of outputting query results as they happen, you can use the SLIDE operator in a subclause to batch the output events. You can batch the events based on the number of events when you use the ROW operator or an amount of time (time window) when you use the RANGE operator.

SELECT * FROM StockTradeIStreamChannel[ROWS 3 SLIDE 2]

RSTREAM(SELECT * FROM StockTradeIStreamChannel[ROWS 3 SLIDE 2])

<processor>
  <rules>
    <query id=rule2 <![CDATA[RSTREAM (SELECT *
       FROM StockTradeIStreamChannel [ROWS 3 SLIDE 2 >
    </query>
  </rules>
</processor>

SELECT * FROM StockTradeIStream[RANGE 5 MIN SLIDE 30 SECONDS]

<processor>
  <rules>
    <query id=rule2 <![CDATA[RSTREAM (SELECT *
       FROM StockTradeIStreamChannel [RANGE 5 MIN 30 SECONDS >
    </query>
  </rules>
</processor>

Views enable you to create an Oracle CQL statement that can be reused by other Oracle CQL queries. A view element contains Oracle CQL subquery statements. The view element id attribute defines the view name. A top-level SELECT statement that you create in a view element is called a view.

The following example shows view v1 and query q1 on the view. The view selects from stream S1 of xmltype stream elements. The view v1 uses the XMLTABLE clause to parse data from the xmltype stream elements with XPath expressions. The query q1 selects from view v1 as it would from any other data source. The XMLTABLE clause also supports XML name spaces.

An xmltype stream contains XML data. With the Oracle CQL XMLTABLE clause, you can parse data from an xmltype stream into columns using XPath expressions and access the data by column name. XPath expressions enable you to navigate through elements and attributes in an XML document.

<view id="v1" schema="orderId LastShares LastPrice"><![CDATA[
  SELECT
    X.OrderId,
    X.LastShares,
    X.LastPrice
  FROM S1, XMLTABLE (
   "FILL"
   PASSING BY VALUE
   S1.c1 as "."
   COLUMNS
      OrderId char(16) PATH "fn:data(../@ID)",
      LastShares integer PATH "fn:data(@LastShares)",
      LastPrice float PATH "fn:data(@LastPx)"
   ) as X
></view>

<query id="q1"><![CDATA[
  IStream(
    select
      orderId,
      sum(LastShares * LastPrice),
      sum(LastShares * LastPrice) / sum(LastShares)
    from
      v1[now]
    group by orderId
  )
></query>

The following assembly file entry shows the setting for a table source with an id attribute of Stock.

<wlevs:table id="Stock" event-type="TradeEvent" data-source="StockDataSource"/>

Oracle Stream Analytics uses the event type and the data source to map a relational table row to the event type. The TradeEvent event type is created from a Java class that has the following five private fields that map to columns in the relational database: symbol, price, lastPrice, percChange, and volume.

While specifying support for parallel query execution is at its core a simple configuration task, be sure to follow the other steps below so that you get the most out of the feature.

• Use the ordering-constraint attribute to support parallel execution.

• Make sure you have enough threads calling into the processor to meet your performance goals. The maximum amount of parallel query execution is constrained by the number of threads available to the CQL processor. For example, if an adapter upstream of the processor supports the number of threads you need and there is a channel between the adapter and the processor, try configuring the channel with a max-threads count of 0 so that it acts as a pass-through.

  If you don't want a pass-through, be sure to configure the query's upstream channel with a max-threads value greater than 1. (To make a max-threads value setting useful, you'll need to also set the max-size attribute to a value greater than 0.) For more information, see Channels .

• Follow other guidelines related to setting the max-threads attribute value. For example, to make a max-threads value setting useful, you'll need to also set the max-size attribute to a value greater than 0.

• Ensure, if necessary, that a bean receiving the query results is thread-aware, such as by using synchronized blocks. For example, you might need to do so if the bean's code builds a list from results received from queries executed on multiple threads.

You enable parallel query execution by relaxing the default ordering constraint that ensures that events are processed serially. You do this by setting the ordering-constraint attribute on a query or view element.

In the following example, the ordering-constraint attribute is set to UNORDERED so that the query will execute in parallel whenever possible:

<query id="myquery" ordering-constraint="UNORDERED">
    SELECT symbol FROM S WHERE price > 10
</query>

The ordering-constraint attribute supports the following three values:

• ORDERED means that the order of output events (as implied by the order of input events) is important. The CQL engine will process events serially. This is the default behavior.

• UNORDERED means that order of the output events is not important to the consumer of the output events. This gives the freedom to the CQLProcessor to process events in parallel on multiple threads. When possible, the query will execute in parallel on multiple threads to process the events.

• PARTITION_ORDERED means that you're specifying that order of output events within a partition is to be preserved (as implied by the order of input events) while order of output events across different partitions is not important to the consumer of the output events. This relaxation provides some freedom to the CQL engine to process events across partitions in parallel (when possible) on multiple threads.

Use the PARTITION_ORDERED value when you want to specify that events conforming to a given partition are processed serially, but that order can be disregarded across partitions and events belonging to different partitions may be processed in parallel. When using the PARTITION_ORDERED value, you must also add the partition-expression attribute to specify which expression for partitioning should be the basis for relaxing the cross-partition ordering constraint.

In the following example, the GROUP BY clause partitions the output based on symbol values. The partition-expression attribute specifies that events in a given subset of events corresponding to a particular symbol value should be handled serially. Across partitions, on the other hand, order can be disregarded.

<query id="myquery" ordering-constraint="PARTITION_ORDERED"
    partitioning-expression="symbol">
    SELECT
        COUNT(*) as c, symbol
    FROM
        S[RANGE 1 minute]
    GROUP BY
        symbol
</query>

In general, you will probably see improved performance for queries by making more threads available and setting the ordering-constraint attribute so that they're able to execute in parallel when possible. As with most performance tuning techniques, a little trial and error with these settings should yield a combination that gets better results.

However, in some cases where your queries use partitioning -- and you've set the ordering-constraint attribute to PARTITION_ORDERED -- you might not see the amount of scaling you'd expect. For example, consider a case in which running with four threads doesn't improve performance very much over running with two threads. In such a case, you can try using the partition-order-capacity value to get the most out of CQL engine characteristics at work with queries that include partitions.

The partition-order-capacity value specifies the maximum amount of parallelism that will be permitted within a given processor instance when processing a PARTITION_ORDERED query. When available threads are handling events belonging to different partitions, the value sets a maximum number of threads that will be allowed to simultaneously run in the query.

As with other aspects of performance tuning, getting the most out of partition-order-capacity may take a bit of experimentation. When tuning with partition-order-capacity, a good starting point is to set it equal to the maximum number of threads you expect to have active in any CQL processor instance. In some cases (for example, at high data rates or with expensive processing downstream from the CQL processor), it may be helpful to set the partition-order-capacity value even higher than the available number of threads. However, you should only do this if performance testing confirms that it's helpful for a given application and load.

The partition-order-capacity value is set from one of four places, two of which you can fall back on when you do not explicitly set it yourself. For information about the settings, see Schema Reference for Oracle Stream Analytics.

These are, in order of precedence.

1. The partition-order-capacity element set on a channel configuration. If you specify this on the input channel for a processor, it takes effect for any PARTITION_ORDERED queries in that processor.

2. The partition-order-capacity property in server configuration. This value will be used for all PARTITION_ORDERED queries running on the server unless the value is set on a channel.

3. The max-threads value set on a channel configuration. If you specify this on the input channel for a processor, it takes effect for any PARTITION_ORDERED queries in that processor

4. A system default value (currently set to 4) is used if you don't specify either a partition-order-capacity value or max-threads value, or if the max-threads value is set to 0 (meaning it's a pass-through channel).

When using partition-order-capacity, keep in mind the following:

• The partition-order-capacity value is only useful when you're setting the ordering-constraint attribute to PARTITION_ORDERED.

• Increasing partition-order-capacity generally increases parallelism and scaling. For example, if your profiling reveals lock contention bottlenecks, you might find it helpful to increase partition-order-capacity to see if contention is reduced.

• Setting partition-order-capacity even higher than the number of available threads can be helpful in some cases because of the particular way partitioning is done in the CQL processor.

• There is some resource cost in memory used by specifying very high values.

• Tuning this parameter is very dependent on details of the application and the input rate. Tuning by experimentation may be necessary to determine an optimal value.

Think of parallel query execution as a performance enhancement feature that you specify support for so that the CQL processor can use it whenever possible. Not all queries can be executed in parallel. This includes queries using certain CQL language features.

For example, if your query uses some form of aggregation -- such as to find the maximum value from a range of values -- the CQL processor may not be able to fully execute the query in parallel (this is needed to guarantee the correct result considering the ordering constraint). Some query semantics in themselves also constrain the query to ordered processing. Such queries will be executed serially regardless of whether you specify support for parallel execution.

Also, the IStream, RStream and DStream operators maintain the state of their operand for processing, making it necessary for the CQL processor to synchronize threads in order to execute the query.

Note that the CQL processor always respects the semantic intention of your query. In cases where the ordering-constraint attribute would change this intention, the attribute is coerced to a value that keeps the intention intact.

If you're using the partitioning-expression attribute, keep in mind that the attribute supports a single expression only. Entering multiple property names for the value is not supported.

You create a fault handler class by implementing the com.bea.wlevs.ede.api.FaultHandler interface. After you have written the class, you associate it with the stage for which it handles faults by registering it as an OSGi service. For more information, see Register a Fault Handler.

Your implementation of the handleFault method receives exceptions for the EPN stage with which the handler is associated. The exception itself is either an instance of com.bea.wlevs.ede.api.EventProcessingException or, if there has been a JVM error, an instance of java.lang.Error.

The method also receives a string array that contains the names of upstream stages, or catchers, to which the exception goes when your code rethrows it. If there is more than one catcher in the array, your rethrown exception goes to all of them. There are two cases when the catchers array is empty: when the exception occurs while executing a temporal query and when the exception is thrown to a channel's fault handler. In these cases, the fault handler executes in the context of a background thread, and there is no linkage to upstream stages.

An exception that is rethrown from a fault handler travels through upstream EPN stages until it is either caught or reaches a stage that cannot catch it (such as a processor or multithreaded channel that does not have an associated fault handler). Note that if you rethrow an exception, any channels in the catcher's list must have an associated fault handler to catch the exception.

The EventProcessingException instance could also be one of the exception types that extend that class, including CQLExecutionException, ArithmeticExecutionException, and others. See the Java API Reference for Oracle Stream Analytics. The EventProcessingException instance provides methods with which your code can retrieve insert, delete, and update events that were involved in generating the fault.

Your implementation of the method should do one of the following:

• Consume the fault in the way that a Java try and catch statement might. If your implementation does not rethrow the fault, then event processing continues with subsequent events. However, query processing continues with its state reset as if the query had been restarted. The processing state is lost and processing begins fresh with events that follow those that provoked the fault.

• Rethrow the fault so that it is received by upstream stages (or their fault handlers). As when the fault is consumed, queries continue processing events, although the query state is reset with subsequent events. The upstream stage receiving the fault always has the option of explicitly stopping the offending query by using the CQL processor's MBean interface.

  Note:

  When you update an Oracle CQL query with an MBean, do not send events during the update procedure. If you send events during some queries, the order of the events in the output stream is not guaranteed. For example, when you update an Oracle CQL query from unordered to ordered in an Oracle CQL parallelism execution.

In the following example the code provides a high-level illustration of handling a fault.

package com.example.faulthandler;

import com.bea.wlevs.ede.api.FaultHandler;

public class SimpleFaultHandler implements FaultHandler
{
    private String suppress;

    // Called by the server to pass in fault information.
    @Override
    public void handleFault(Throwable fault, String[] catchers) throws Throwable
    {
        // Log the fault.
        return;
    }
}

After you have written a fault handling class, you can associate it with an EPN stage by registering it as an OSGi service. The simplest way to do this is to register the handler declaratively in the EPN assembly file.

In the following example, the EPN assembly file excerpt shows a service element stanza that registers the SimpleFaultHandler class as the fault handler for the Oracle CQL processor with an id of exampleProcessor.

<osgi:service interface="com.bea.wlevs.ede.api.FaultHandler">
    <osgi:service-properties>
        <entry key="application.identity" value="myapp"/>
        <entry key="stage.identity" value="exampleProcessor"/>
    </osgi:service-properties>
    <bean class="com.example.faulthandler.SimpleFaultHandler"/>
</osgi:service>

<!-- A processor with a user-defined function. -->
<wlevs:processor id="exampleProcessor" >
    ...
</wlevs:processor>

For more on the schema for registering OSGi services, see http://static.springsource.org/osgi/docs/1.1.x/reference/html/appendix-schema.html. For more on OSGi, see http://en.wikipedia.org/wiki/OSGi.