DYNAMIC PARALLEL AGGREGATION WITH HYBRID BATCH FLUSHING

• US 20140156636A1
• Filed: 12/04/2012
• Published: 06/05/2014
• Est. Priority Date: 12/04/2012
• Status: Active Grant

• Current Assignee: Oracle International Corporation (Oracle Corporation)
• Original Assignee: Oracle International Corporation (Oracle Corporation)
• Inventors: Bellamkonda, Srikanth, Li, Huagang
• Granted Patent: US 9,460,154 B2
• Time in Patent Office: Days
• US Class Current: 707/718
• CPC Class Codes: G06F 16/24556 Aggregation; Duplicate elimination

Abstract

A method, apparatus, and system for dynamic parallel aggregation with hybrid batch flushing are provided. Record sources of an aggregation operator in a query execution plan may dynamically aggregate using the same aggregation operator. The dynamic aggregation creates a batch of aggregation records from an input source, which are then used to aggregate further records from the input source. If a record from the input source is not matched to an aggregation record in the batch, then the record is passed to the next operator. In this manner, records are aggregated ahead of time at a record source to reduce the number of records passed between operators, reducing the impact of network I/O between nodes of a parallel processing system. By adjusting the contents of the batch according to aggregation performance monitored during run-time, hybrid batch flushing can be implemented to adapt to changing data patterns and skewed values.

Claims

1. A method, comprising steps of:

executing a query execution plan, said query execution plan including a first aggregation operator and a second aggregation operator that is a record source for said first aggregation operator, wherein said first aggregation operator and said second aggregation operator perform the same aggregation operation, wherein executing the query execution plan comprises executing the second aggregation operator, wherein executing the second aggregation operator comprises;

generating an initial version of a batch of aggregation records;

after generating said initial version of said batch of aggregation records, determining whether a first record may be aggregated with any aggregation record in said batch of aggregation records; and

in response to determining that said first record may not be aggregated with any aggregation record of said batch of aggregation records, sending said first record to said first aggregation operator;

wherein the method is performed by one or more computing devices.

2. The method of claim 1, wherein the executing of the second aggregation operator further comprises:

monitoring a number of records reduced by aggregating with said initial version of said batch of aggregation records;

determining that said number of records reduced is under a performance threshold;

flushing said initial version of said batch of aggregation records to said first aggregation operator; and

generating a refreshed version of said batch of aggregation records.

3. The method of claim 2, wherein said number of records reduced is calculated as an average over time.

4. The method of claim 2, wherein the executing of the second aggregation operator further comprises:

monitoring a number of successive occurrences of said flushing; and

disabling aggregation at said second aggregation operator in response to said number of successive occurrences exceeding a set number.

5. The method of claim 2, wherein said flushing reserves a portion of said initial version of said batch of aggregation records, and wherein said generating of said refreshed version of said batch of aggregation records preserves said reserved portion.

6. The method of claim 1, wherein said first aggregation operator is performed by a first plurality of said one or more computing devices, and wherein said second aggregation operator is performed by a second plurality of said one or more computing devices.

7. The method of claim 1, wherein said generating of said initial version of said batch of aggregation records is from an input source of said second aggregation operator.

8. The method of claim 7, wherein said generating uses a frequency of keys of said same aggregation operation in said input source to populate said initial version of said batch of aggregation records.

9. The method of claim 1, wherein said initial version of said batch of aggregation records is limited to a maximum number of records.

10. The method of claim 9, wherein said maximum number of records is based on available memory of said one or more computing devices.

11. The method of claim 9, wherein said maximum number of records is based on available processor cache of said one or more computing devices.

12. The method of claim 1, wherein said initial version of said batch of aggregation records is stored in a hash table.

13. A non-transitory computer-readable medium storing one or more sequences of instructions which, when executed by one or more processors, cause performing of:

executing a query execution plan, said query execution plan including a first aggregation operator and a second aggregation operator that is a record source for said first aggregation operator, wherein said first aggregation operator and said second aggregation operator perform the same aggregation operation, wherein executing the query execution plan comprises executing the second aggregation operator, wherein executing the second aggregation operator comprises;

generating an initial version of a batch of aggregation records;

after generating said initial version of said batch of aggregation records, determining whether a first record may be aggregated with any aggregation record in said batch of aggregation records; and

in response to determining that said first record may not be aggregated with any aggregation record of said batch of aggregation records, sending said first record to said first aggregation operator.

14. The non-transitory computer-readable medium of claim 13, wherein the executing of the second aggregation operator further comprises:

monitoring a number of records reduced by aggregating with said initial version of said batch of aggregation records;

determining that said number of records reduced is under a performance threshold;

flushing said initial version of said batch of aggregation records to said first aggregation operator; and

generating a refreshed version of said batch of aggregation records.

15. The non-transitory computer-readable medium of claim 14, wherein said number of records reduced is calculated as an average over time.

16. The non-transitory computer-readable medium of claim 14, wherein the executing of the second aggregation operator further comprises:

monitoring a number of successive occurrences of said flushing; and

disabling aggregation at said second aggregation operator in response to said number of successive occurrences exceeding a set number.

17. The non-transitory computer-readable medium of claim 14, wherein said flushing reserves a portion of said initial version of said batch of aggregation records, and wherein said generating of said refreshed version of said batch of aggregation records preserves said reserved portion.

18. The non-transitory computer-readable medium of claim 13, wherein said first aggregation operator is performed by a first plurality of computing devices, and wherein said second aggregation operator is performed by a second plurality of computing devices.

19. A database system comprising:

a network;

a query execution plan including a first aggregation operator and a second aggregation operator that is a record source for said first aggregation operator, wherein said first aggregation operator and said second aggregation operator perform the same aggregation operation, wherein executing the query execution plan comprises executing the second aggregation operator;

a first plurality of computing devices executing the first aggregation operator;

a second plurality of computing devices executing the second aggregation operator comprising;

generating an initial version of a batch of aggregation records;

after generating said initial version of said batch of aggregation records, determining whether a first record may be aggregated with any aggregation record in said batch of aggregation records; and

in response to determining that said first record may not be aggregated with any aggregation record of said batch of aggregation records, sending said first record to said first aggregation operator over said network.

20. The database system of claim 19 further comprising a database storage subsystem, wherein said first plurality of computing devices are database servers, and wherein said second plurality of computing devices are database servers in communication with said database storage subsystem.