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

• US 8,521,867 B2
• Filed: 10/20/2007
• Issued: 08/27/2013
• Est. Priority Date: 10/20/2007
• Status: Active Grant

• Current Assignee: Oracle International Corporation (Oracle Corporation)
• Original Assignee: Oracle International Corporation (Oracle Corporation)
• Inventors: Srinivasan, Anand, Jain, Namit, Mishra, Shailendra Kumar
• Primary Examiner(s): Backer, Firmin
• Assistant Examiner(s): EDWARDS, JAMES A
• Application Number:
  US11/977,437
• Publication Number:
  US 20090106440A1
• Time in Patent Office: 2,138 Days
• Field of Search: 709/224
• US Class Current: 709/224
• 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 for 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;

receiving a command to register a user-defined aggregation and an identification of a set of instructions comprising a function to be executed to perform the user-defined aggregation over multiple tuples of the data in messages received from the plurality of streams;

creating in a memory of the computer, a first structure comprising said identification of the set of instructions;

receiving a new continuous query comprising the user-defined aggregation;

during execution of the new continuous query;

creating in the memory, an instance of said function to be executed to perform the user-defined aggregation, by using the identification in the first structure in response to receipt of a message from the plurality of streams;

invoking within said instance created in response to receipt of the message, the function to perform the user-defined aggregation to process a tuple of the data in the message;

continuing to execute the new continuous query, by performing the invoking and skipping the creating, in response to receipt of an additional message from the plurality of streams; and

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

2. The method of claim 1 further comprising:

receiving with the command, an additional identification of a value of a data type of at least one argument of the user-defined aggregation;

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, an additional identification of a class containing the function, and a name of a package containing the class;

wherein the command comprises a clause providing the additional identification; and

wherein 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 user-defined aggregation;

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

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

5. The method of claim 1 wherein:

the user-defined 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 method further comprising creating an additional instance of the set of instructions for operation on data of a second value of the attribute.

6. The method of claim 1 wherein:

the function to be executed to perform the user-defined aggregation is identified from among multiple functions based at least on a first type of the message;

the instance of the function to be executed to perform the user-defined aggregation is created when a count is at an initial value on receipt of the message; and

the method further comprises;

changing the count, by one of incrementing or decrementing, to obtain a new value in response to receipt of the message;

changing the count by the other of incrementing or decrementing, on receipt from the plurality of streams of another message of a second type different from the first type; and

releasing memory occupied by the instance and changing the count to the initial value when the count is at the new value on receipt of yet another message of the second type from the plurality of streams.

7. The method of claim 6 wherein:

the first type indicates addition of the tuple into a window, the second type indicates removal of the tuple from the window, and the count indicates a number of tuples in the window; and

a store is not maintained, of tuples currently in the window.

8. The method of claim 1 wherein:

the tuple in the message is processed only once to perform the user-defined aggregation, by use of state variables stored internally by the function in said instance and passed between successive invocations of the function.

9. One or more non-transitory computer-readable storage media comprising a plurality of instructions to be executed by a computer processing a plurality of streams of data, the instructions comprising:

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

instructions to receive;

(A) a command to register a user-defined aggregation; and

(B) an identification of a set of instructions comprising a function, to be executed to perform the user-defined aggregation over multiple tuples of the data in messages received from the plurality of streams;

instructions to create in a computer memory, a first structure comprising said identification;

instructions to receive a new continuous query comprising the user-defined aggregation;

instructions to create in the memory, an instance of said function to be executed to perform the user-defined aggregation, by using the identification in the first structure in response to receipt of a message from the plurality of streams;

instructions to invoke within the instance created in response to receipt of the message, the function to perform the user-defined aggregation to process a tuple of data in the message;

instructions to continue to execute the new continuous query, by executing the instructions to invoke and skipping the instructions to create, in response to receipt of an additional message from the plurality of streams; and

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

10. The one or more non-transitory computer-readable storage media of claim 9 further comprising:

instructions to receive with the command, an additional identification of a value of a data type of at least one argument.

11. The one or more non-transitory computer-readable storage media of claim 10 wherein:

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

12. The one or more non-transitory computer-readable storage media of claim 9 further comprising:

instructions to receive with the command, an additional identification of a class containing the function, and a name of a package containing the class.

13. The one or more non-transitory computer-readable storage media of claim 12 wherein:

the command comprises a clause providing the additional identification; and

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

14. The one or more non-transitory computer-readable storage media of claim 9 wherein:

the first structure further comprises a name of the user-defined aggregation.

15. The one or more non-transitory computer-readable storage media of claim 9 wherein:

the first structure further comprises at least one data type of an argument of the user-defined aggregation.

16. The one or more non-transitory computer-readable storage media of claim 9 wherein:

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

17. The one or more non-transitory computer-readable storage media of claim 9 wherein:

the user-defined 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; and

the one or more non-transitory computer-readable storage media further comprise instructions to create an additional instance of the set of instructions for operation on data of a second value of the attribute.

18. The one or more non-transitory computer-readable storage media of claim 9 wherein:

the function to be executed to perform the user-defined aggregation is identified from among multiple functions based at least on a first type of the message;

the instance of the function to be executed to perform the user-defined aggregation is created when a count is at an initial value on receipt of the message;

the one or more non-transitory computer-readable storage media further comprise;

instructions to change the count, by one of incrementing or decrementing, to obtain a new value in response to receipt of the message;

instructions to change the count by the other of incrementing or decrementing, on receipt from the plurality of streams of another message of a second type different from the first type; and

instructions to release memory occupied by the instance and change the count to the initial value, when the count is at the new value on receipt of yet another message of the second type from the plurality of streams.

19. The one or more non-transitory computer-readable storage media of claim 18 wherein:

the first type indicates addition of the tuple into a window, the second type indicates removal of the tuple from the window, and the count indicates a number of tuples in the window; and

during execution of said instructions, a store is not maintained, of tuples currently in the window.

20. The one or more non-transitory computer-readable storage media of claim 9 wherein:

the tuple in the message is processed only once to perform the user-defined aggregation, by use of state variables stored internally by the function in said instance and passed between successive invocations of the function.

21. An apparatus for processing a plurality of streams of data, the apparatus comprising:

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

means, coupled to the means for processing, for receiving a command to register a user-defined aggregation and an identification of a set of instructions comprising a function to be executed to perform the user-defined aggregation over multiple tuples of the data in messages received from the plurality of streams;

means, coupled to the means for processing, for creating in a memory a first structure comprising said identification of the set of instructions;

means, coupled to the means for processing, for receiving a new continuous query comprising the user-defined aggregation;

means for creating an instance of the function to be executed to perform the user-defined aggregation, by using the identification in the first structure in response to receipt of a message from the plurality of streams;

means for invoking within said instance created in response to receipt of the message, the function to perform the user-defined aggregation to process a tuple of the data in the message;

means for continuing to execute the new continuous query, by operating the means for the invoking and skipping operation of the means for creating, in response to receipt of an additional message from the plurality of streams; and

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

22. The apparatus of claim 21 further comprising:

means for receiving an identification of a value of a data type of 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.

23. The apparatus of claim 21 further comprising:

means for receiving an identification of a class containing the function, and a name of a package containing the class;

wherein the command comprises a clause providing the identification; and

wherein the set is identified by a predetermined label within the package.

24. The apparatus of claim 21 wherein:

the first structure further comprises a name of the user-defined aggregation;

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

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

25. The apparatus of claim 21 wherein:

the user-defined 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; and

the apparatus further comprises means for creating an additional instance of the set of instructions for operation on data of a second value of the attribute.