Support for incrementally processing user defined aggregations in a data stream management system

• US 20090106440A1
• Filed: 10/20/2007
• Published: 04/23/2009
• Est. Priority Date: 10/20/2007
• Status: Active Grant

• Current Assignee: Oracle International Corporation (Oracle Corporation)
• Original Assignee: Oracle International Corporation (Oracle Corporation)
• Inventors: Jain, Namit, Srinivasan, Anand, Mishra, Shailendra Kumar
• Granted Patent: US 8,521,867 B2
• Time in Patent Office: Days
• US Class Current: 709/231
• CPC Class Codes: G06F 16/244 Grouping and aggregation
• CPC Class Codes: G06F 16/24568 Data stream processing; Continuous queries

Abstract

A computer is programmed to accept a command for creation of a new aggregation defined by a user to process data incrementally, one tuple at a time. One or more incremental function(s) in a set of instructions written by the user to implement the new aggregation maintain(s) locally any information that is to be passed between successive invocations, to support computing the aggregation for a given set of tuples as a whole. The user writes a set of instructions to perform the aggregation incrementally, including a plus function which is repeatedly invoked, only once, for each addition to a window of a message. The user also writes a minus function to be invoked with the message, to return the value of incremental aggregation over the window after removal of the message. In such embodiments, the computer does not maintain copies of messages in the window for use by aggregation function(s).

Claims

1. A method implemented in a computer of processing a plurality of streams of data, the method comprising:

processing the plurality of streams, to execute thereon a plurality of continuous queries based on a global plan;

during the processing, receiving a command to create an aggregation and identification of a set of instructions comprising a function to be executed to perform the aggregation;

during the processing, creating in a memory of the computer, a first structure comprising the identification;

during the processing, receiving a new continuous query to be executed using the aggregation;

during the processing, based on the first structure, creating in the memory an operator comprising at least one count and at least one second structure, the count being initialized to an initial value, the second structure comprising a first field to hold a reference to the instance, and at least one additional field corresponding to at least one argument of the aggregation;

during the processing, modifying the global plan by adding thereto the operator, thereby to obtain a modified plan; and

altering the processing, to cause execution of the new continuous query in addition to the plurality of continuous queries, based on the modified plan;

during execution of the new continuous query;

if the count is at the initial value on receipt of a message, creating an instance of the set of instructions;

changing the count to a new value;

invoking the function in the instance to process a tuple of the data in the message, wherein the function is identified based at least on a first type of the message;

repeatedly performing the changing and the invoking, for each receipt of an additional message;

wherein if the additional message is of the first type, the invoking is performed only once; and

if the count is at the new value on receipt of yet another message of a second type, releasing memory occupied by the instance and changing the count to the initial value;

wherein the first type indicates addition of the message into a window and the second type indicates removal of the message from the window; and

outputting from said computer, a stream generated based at least partially on processing of said data by executing the new continuous query.

2. The method of claim 1 further comprising:

receiving with the command, identification of a value of a data type of the at least one argument;

wherein the function is further identified, for use in the invoking, based at least on the value of the data type.

3. The method of claim 1 further comprising:

receiving with the command, identification of a class containing the function, and a name of a package containing the class;

the command comprises a clause providing the identification; and

the set of instructions is identified by a predetermined label within the package.

4. The method of claim 1 wherein:

the first structure further comprises a name of the aggregation;

the first structure further comprises at least one data type of an argument of the aggregation; and

the first structure further comprises at least another data type of value to be returned by the aggregation.

5. The method of claim 1 wherein:

the aggregation groups data in the plurality of streams by an attribute of the data specified in the command;

the instance operates only on data of a first value of the attribute;

the count is changed only for data of the first value;

the operator comprising a plurality of additional counts and a plurality of additional second structures, for a corresponding plurality of additional values of the attribute; and

the method further comprising

if an additional count is at the initial value on receipt of a message of the first kind for an additional value of the attribute, creating an additional instance of the set of instructions and changing the additional count to the new value; and

if the additional count is at the new value on receipt of another message of the second kind for the additional value of the attribute, releasing memory occupied by the additional instance and changing the additional count to the initial value.