Support for user defined aggregations in a data stream management system

• US 7,991,766 B2
• Filed: 10/20/2007
• Issued: 08/02/2011
• Est. Priority Date: 10/20/2007
• Status: Active Grant

• Current Assignee: Oracle International Corporation (Oracle Corporation)
• Original Assignee: Oracle International Corporation (Oracle Corporation)
• Inventors: Mishra, Shailendra Kumar, Jain, Namit, Srinivasan, Anand
• Primary Examiner(s): Vy; Hung T
• Application Number:
  US11/977,439
• Publication Number:
  US 20090106218A1
• Time in Patent Office: 1,382 Days
• Field of Search: 707/713, 707/714, 707/715, 707/718, 709/231
• US Class Current: 707/714
• CPC Class Codes: G06F 16/24556 Aggregation; Duplicate elimination
• CPC Class Codes: G06F 16/24568 Data stream processing; Continuous queries

Abstract

A computer is programmed to accept a command to create a new aggregation defined by a user during execution of continuous queries on streams of data. The computer is further programmed to thereafter accept and process new continuous queries using the new aggregation, in a manner similar to built-in aggregations. The user typically writes a set of instructions to perform the new aggregation, and identifies in the command, a location of the set of instructions. In response to such a command, the computer creates metadata identifying the new aggregation. The metadata is used to instantiate one aggregation for each group of data in a current window, grouped by an attribute identified in a new query.

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 said processing, receiving a command to create an aggregation and identification of a set of instructions to be executed to perform said aggregation;

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

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

during said processing, based on said first structure, creating in said memory an operator comprising at least one second structure, the second structure comprising a first field to hold a reference to an instance of said set of instructions, and at least one additional field to hold information on at least one argument of said aggregation;

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

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

during said execution of the new continuous query, creating said instance of said set of instructions;

during said execution of the new continuous query, using said instance to perform said aggregation on at least a portion of said data; 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 said command, identification of a type of said at least one argument;

wherein said creation of said instance is based on said type.

3. The method of claim 1 further comprising:

receiving with said command, identification of a class comprising said aggregation, and a name of a package containing said class;

wherein the command comprises a clause, said clause comprising a reserved word and said identification; and

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

4. The method of claim 1 wherein:

said first structure further comprises a name of said aggregation;

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

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

5. The method of claim 1 further comprising:

checking to ensure said aggregation is identified in said new continuous query in a select list.

6. The method of claim 1 wherein:

said second structure further comprises a data type of a result to be returned by said aggregation.

7. The method of claim 1 wherein:

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

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

a count corresponds to said first value of said attribute;

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

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

said method further comprising;

creating an additional instance of said set of instructions; and

using said additional instance to process at least another portion of said data.

8. The method of claim 1 wherein:

said second structure further comprises a second field, and the method further comprises copying into said second field a result returned by said aggregation.

9. One or more non-transitory computer readable storage media comprising instructions that when executed by a computer cause the computer to:

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

during execution of instructions to process, receive a command to create an aggregation and identification of a set of instructions to be executed to perform said aggregation;

during execution of instructions to process, create in a memory of said computer, a first structure comprising said identification;

during execution of instructions to process, receive a new continuous query to be executed using said aggregation;

during execution of instructions to process, based on said first structure, create in said memory an operator comprising at least one second structure, the second structure comprising a first field to hold a reference to an instance of said set of instructions, and at least one additional field to hold information on at least one argument of said aggregation;

during execution of instructions to process, modify the global plan by adding thereto said operator, thereby to obtain a modified plan;

alter said instructions to process, to cause execution of the new continuous query in addition to said plurality of continuous queries, based on the modified plan;

during said execution of the new continuous query, create said instance of said set of instructions;

during said execution of the new continuous query, use said instance to perform said aggregation on at least a portion of said data; and

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

10. The one or more non-transitory computer readable storage media of claim 9 wherein execution of said instructions further cause said computer to:

receive with said command, identification of a type of said at least one argument;

wherein said instance of said set of instructions is created based on said type.

11. The one or more non-transitory computer readable storage media of claim 9 wherein execution of said instructions further cause said computer to:

receive with said command, identification of a class comprising said aggregation, and a name of a package containing said class;

wherein the command comprises a clause, said clause comprising a reserved word and said identification.

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

said first structure further comprises a name of said aggregation;

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

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

13. The one or more non-transitory computer readable storage media of claim 9 wherein execution of said instructions further cause said computer to:

check to ensure said aggregation is identified in said new continuous query in a select list.

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

said second structure further comprises a data type of a result to be returned by said aggregation.

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

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

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

a count corresponds to said first value of said attribute;

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

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

execution of said instructions further cause said computer to;

create an additional instance of said set of instructions; and

use said additional instance to process at least another portion of said data.

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

said second structure further comprises a second field, and execution of said instructions further cause said computer to copy into said second field a result returned by said aggregation.

17. 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 for receiving a command to create an aggregation and identification of a set of instructions to be executed to perform said aggregation;

means for creating in a memory, a first structure comprising said identification;

means for receiving a new continuous query to be executed using said aggregation;

means, based on said first structure, for creating in said memory an operator comprising at least one second structure, the second structure comprising a first field to hold a reference to an instance of said set of instructions, and at least one additional field to hold information on at least one argument of said aggregation;

means for modifying the global plan by adding thereto said operator, thereby to obtain a modified plan;

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

means for creating said instance of said set of instructions;

means for using said instance to perform said aggregation on at least a portion of said data; and

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

18. The apparatus of claim 17 further comprising:

means for checking to ensure said aggregation is identified in said new continuous query in a select list.

19. The apparatus of claim 17 wherein:

said second structure further comprises a data type of a result to be returned by said aggregation.

20. The apparatus of claim 17 wherein:

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

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

a count corresponds to said first value of said attribute;

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

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

said apparatus further comprising;

means for creating an additional instance of said set of instructions; and

means for using said additional instance to process at least another portion of said data.