10 Functions: User-Defined

10.1.1 Types of User-Defined Functions

Using the classes in the oracle.cep.extensibility.functions package you can create the following types of user-defined functions:

• Single-row: a function that returns a single result row for every row of a queried stream or view (for example, like the CONCAT built-in function does).

  For more information, see "How to Implement a User-Defined Single-Row Function".

• Aggregate: a function that implements com.bea.wlevs.processor.AggregationFunctionFactory and returns a single aggregate result based on group of tuples, rather than on a single tuple (for example, like the SUM built-in function does).

  Consider implementing your aggregate function so that it performs incremental processing, if possible. This will improve scalability and performance because the cost of (re)computation on arrival of new events will be proportional to the number of new events as opposed to the total number of events seen thus far.

  For more information, see "How to Implement a User-Defined Aggregate Function".

You can create overloaded functions and you can override built-in functions.

10.1.2 User-Defined Function Datatypes

Table 10-1 lists the datatypes you can specify when you implement and register a user-defined function.

The Oracle CQL Datatype column lists the datatypes you can specify in the Oracle CQL statement you use to register your user-defined function and the Equivalent Java Datatype column lists the Java datatype equivalents you can use in your user-defined function implementation.

At run time, Oracle CEP maps between the Oracle CQL datatype and the Java datatype. If your user-defined function returns a datatype that is not in this list, Oracle CEP will throw a ClassCastException.

For more information about data conversion, see Section 2.3.4, "Data Conversion".

10.1.3 User-Defined Functions and the Oracle CEP Server Cache

You can access an Oracle CEP cache from an Oracle CQL statement or user-defined function.

For more information, see:

• "Configuring Oracle CEP Caching" in the Oracle CEP IDE Developer's Guide for Eclipse

• "Accessing a Cache From an Oracle CQL Statement" in the Oracle CEP IDE Developer's Guide for Eclipse

• "Accessing a Cache From an Oracle CQL User-Defined Function" in the Oracle CEP IDE Developer's Guide for Eclipse

10.2.1 How to Implement a User-Defined Single-Row Function

You implement a user-defined single-row function by implementing a Java class that implements the SingleElementFunction interface.

To implement a user-defined single-row function:

1. Implement a Java class as Example 10-1 shows.

   Ensure that the data type of the return value corresponds to the return type in the register function statement as Example 10-1 shows.

   Example 10-1 MyMod.java User-Defined Single-Row Function

   package com.bea.wlevs.example.function;
   
   public class MyMod {
       public Object execute(Object[] args) {
           int arg0 = ((Integer)args[0]).intValue();
           int arg1 = ((Integer)args[1]).intValue();
           return new Integer(arg0 % arg1);
       }
   }
   

2. Compile the user-defined function Java implementation class and register the class in your Oracle CEP application assembly file as Example 10-2 shows.

   Example 10-2 Single-Row User Defined Function for an Oracle CQL Processor

   <wlevs:processor id="testProcessor">
       <wlevs:listener ref="providerCache"/>
       <wlevs:listener ref="outputCache"/>
       <wlevs:cache-source ref="testCache"/>
       <wlevs:function function-name="mymod" exec-method=”execute” />
           <bean class="com.bea.wlevs.example.function.MyMod"/>
       </wlevs:function>
   </wlevs:processor>
   

   For more information, see "wlevs:function" in the Oracle CEP IDE Developer's Guide for Eclipse.

3. Invoke your user-defined function in the select list of a SELECT statement or the condition of a WHERE clause as Example 10-3 shows.

   Example 10-3 Accessing a User-Defined Single-Row Function in Oracle CQL

   ...
   <view id="v1" schema="c1 c2 c3 c4"><![CDATA[ 
       select
           mymod(c1, 100), c2, c3, c4 
       from 
           S1
   ]]></view>
   ...
   <query id="q1"><![CDATA[ 
       select * from v1 [partition by c1 rows 1] where c4 - c3 = 2.3 
   ]]></query>
   ...
   

10.2.2 How to Implement a User-Defined Aggregate Function

You implement a user-defined aggregate function by implementing a Java class that implements the com.bea.wlevs.processor.AggregationFunctionFactory interface.

To implement a user-defined aggregate function:

1. Implement a Java class as Example 10-4 shows.

   Consider implementing your aggregate function so that it performs incremental processing, if possible. This will improve scalability and performance because the cost of (re)computation on arrival of new events will be proportional to the number of new events as opposed to the total number of events seen thus far. The user-defined aggregate function in Example 10-4 supports incremental processing.

   Example 10-4 Variance.java User-Defined Aggregate Function

   package com.bea.wlevs.test.functions;
    
   import com.bea.wlevs.processor.AggregationFunction;
   import com.bea.wlevs.processor.AggregationFunctionFactory;
    
   public class Variance implements AggregationFunctionFactory, AggregationFunction {
    
       private int count;
       private float sum;
       private float sumSquare;
    
       public Class<?>[] getArgumentTypes() {
           return new Class<?>[] {Integer.class};
       }
    
       public Class<?> getReturnType() {
           return Float.class;
       }
    
       public AggregationFunction newAggregationFunction() {
           return new Variance();
       }
    
       public void releaseAggregationFunction(AggregationFunction function) {
       }
    
       public Object handleMinus(Object[] params) {
           if (params != null && params.length == 1) {
               Integer param = (Integer) params[0];
               count--;
               sum -= param;
               sumSquare -= (param * param);
           }
           
           if (count == 0) {
               return null;
           } else {
               return getVariance();
           }
       }
    
       public Object handlePlus(Object[] params) {
           if (params != null && params.length == 1) {
               Integer param = (Integer) params[0];
               count++;
               sum += param;
               sumSquare += (param * param);
           }
           
           if (count == 0) {
               return null;
           } else {
               return getVariance();
           }
       }
    
       public Float getVariance() {
           float avg = sum / (float) count;
           float avgSqr = avg * avg;
           float var = sumSquare / (float)count - avgSqr;
           return var;
       }
    
       public void initialize() {
           count = 0;
           sum = 0.0F;
           sumSquare = 0.0F;
       }
    
   }
   

2. Compile the user-defined function Java implementation class and register the class in your Oracle CEP application assembly file as Example 10-5 shows.

   Example 10-5 Aggregate User Defined Function for an Oracle CQL Processor

   <wlevs:processor id="testProcessor">
        <wlevs:listener ref="providerCache"/>
        <wlevs:listener ref="outputCache"/>
        <wlevs:cache-source ref="testCache"/>
        <wlevs:function function-name="var">
          <bean class="com.bea.wlevs.test.functions.Variance"/>
        </wlevs:function>
      </wlevs:processor>
   

   For more information, see "wlevs:function" in the Oracle CEP IDE Developer's Guide for Eclipse.

3. Invoke your user-defined function in the select list of a SELECT statement or the condition of a WHERE clause as Example 10-6 shows.

   Example 10-6 Accessing a User-Defined Aggregate Function in Oracle CQL

   ...
   <query id="uda6"><![CDATA[ 
       select var(c2) from S4[range 3] 
   ]]></query>
   ...
   

   At run-time, when the user-defined aggregate is executed, and a new event becomes active in the window of interest, the aggregations will have to be recomputed (since the set over which the aggregations are defined has a new member). To do so, Oracle CEP passes only the new event (rather than the entire active set) to the appropriate handler context by invoking the appropriate handlePlus* method in Example 10-4. This state can now be updated to include the new event. Thus, the aggregations have been recomputed in an incremental fashion.

   Similarly, when an event expires from the window of interest, the aggregations will have to be recomputed (since the set over which the aggregations are defined has lost a member). To do so, Oracle CEP passes only the expired event (rather than the entire active set) to the appropriate handler context by invoking the appropriate handleMinus method in Example 10-4. As before, the state in the handler context can be incrementally updated to accommodate expiry of the event in an incremental fashion.