Hash join using collaborative parallel filtering in intelligent storage with offloaded bloom filters

• US 8,825,678 B2
• Filed: 09/18/2009
• Issued: 09/02/2014
• Est. Priority Date: 09/19/2008
• Status: Active Grant

• Current Assignee: Oracle International Corporation (Oracle Corporation)
• Original Assignee: Oracle International Corporation (Oracle Corporation)
• Inventors: Potapov, Dmitry, Lau, Yiu Woon, Jakobsson, Hakan, Panchaksharaiah, Umesh, Kumar, Poojan
• Primary Examiner(s): Trujillo, James
• Assistant Examiner(s): CONYERS, DAWAUNE A
• Application Number:
  US12/562,984
• Publication Number:
  US 20100082648A1
• Time in Patent Office: 1,810 Days
• Field of Search: 707/756, 707/754, 706/20
• US Class Current: 707/754
• CPC Class Codes: G06F 12/00 Accessing, addressing or allocating within memory systems or architectures digital input from, or digital output to record carriers, e.g. to disk storage units, G06F3/06
• CPC Class Codes: G06F 16/2255 Hash tables
• CPC Class Codes: G06F 16/24557 Efficient disk access during query execution
• CPC Class Codes: G06F 16/2456 Join operations
• CPC Class Codes: G06F 3/0611 in relation to response time
• CPC Class Codes: G06F 3/0659 Command handling arrangements, e.g. command buffers, queues, command scheduling
• CPC Class Codes: G06F 3/067 Distributed or networked storage systems, e.g. storage area networks [SAN], network attached storage [NAS]

Abstract

Processing resources at a storage system for a database server are utilized to perform aspects of a join operation that would conventionally be performed by the database server. When requesting a range of data units from a storage system, the database server includes join metadata describing aspects of the join operation for which the data is being requested. The join metadata may be, for instance, a bloom filter. The storage system reads the requested data from disk as normal. However, prior to sending the requested data back to the storage system, the storage system analyzes the raw data based on the join metadata, removing a certain amount of data that is guaranteed to be irrelevant to the join operation. The storage system then returns filtered data to the database server. The database system thereby avoids the unnecessary transfer of certain data between the storage system and the database server.

Claims

1. A method comprising:

in a database system comprising at least a database server and a data storage system that is separate from the database server, performing a join operation between at least a first table and a second table by;

at the database server, generating a bloom filter based upon one or more attributes of the second table;

identifying, at the database server, one or more locations of one or more data blocks in which the data storage system stores data for the first table;

wherein the data storage system comprises one or more storage devices upon which data for tables, including the first table, is stored in a plurality of block structures, including the one or more data blocks;

the database server sending to the data storage system;

a) an input/output (I/O) request identifying the one or more locations of the one or more data blocks stored in the data storage system; and

b) the bloom filter, associated with the I/O request;

wherein the I/O request, is a communication that, when interpreted by the data storage system, causes;

the data storage system reading said one or more data blocks from the one or more storage devices;

the data storage system generating a filtered response to the I/O request from the one or more data blocks by removing data block rows that do not apply to the bloom filter; and

sending to the data storage system metadata indicating to the data storage system one or more data block row fields, of the one or more data blocks, that correspond to one or more particular columns of the first table, wherein removing data block rows that do not apply to the bloom filter comprises comparing results of one or more functions of the values of the one or more data block row fields to the bloom filter;

in response to the I/O request, the database server receiving the filtered response from the data storage system;

performing the join operation based on the filtered response;

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

2. The method of claim 1, further comprising:

prior to sending the bloom filter to the data storage system, the database server receiving a second set of one or more data blocks from the data storage system;

the database server translating the one or more data blocks into the second table;

the database server generating a hash table based upon one or more attributes of a second table;

generating the bloom filter based upon the hash table.

3. The method of claim 2, further comprising:

prior to sending the bloom filter to the data storage system, generating at least one predicate based upon the bloom filter;

sending the bloom filter to the data storage system in the form of the at least one predicate.

4. The method of claim 1, wherein the filtered data includes one or more raw data blocks.

5. The method of claim 1, wherein the I/O request identifies the one or more locations of the one or more data blocks by specifying one or more logical or physical addresses at which the one or more data blocks are stored.

6. The method of claim 1, further comprising:

the database server translating the filtered response into only a partial version of the first table;

the database server performing the join operation based on the partial version of the first table.

7. The method of claim 1, wherein the bloom filter is a first bloom filter, the method further comprising:

at the database server, generating a second bloom filter based upon one or more attributes of a third table that is also involved in the join operation;

the database server sending to the data storage system;

(c) the second bloom filter, also associated with the I/O request;

wherein the I/O request further causes the data storage system generating the filtered response to the I/O request from the one or more data blocks by removing data block rows that do not apply to the second bloom filter.

8. The method of claim 1, further comprising sending the bloom filter in the I/O request.

9. The method of claim 1, further comprising:

sending multiple I/O requests to the data storage system for different sets of one or more data blocks in which the data storage system stores data for the first table, the multiple I/O requests including the I/O request;

wherein at least some of the multiple I/O requests reference a cached copy of the bloom filter stored at the data storage system responsive to the data storage system receiving the bloom filter from the database server.

10. The method of claim 9, wherein the at least some of the multiple I/O requests are from database servers other than the database server.

11. The method of claim 1, further comprising:

sending multiple I/O requests to a plurality of data storage systems, including the data storage system, for different sets of one or more data blocks in which the plurality of data storage systems store data for the first table, the multiple I/O requests including the I/O request;

sending the bloom filter to the data storage system by pushing the bloom filter to a location accessible to all of the plurality of storage systems, and referencing the bloom filter in the I/O request.

12. The method of claim 1, further comprising:

prior to the I/O request, sending the bloom filter to the data storage system in association with a context identifier;

wherein the I/O request includes the context identifier to associate the I/O request with the bloom filter.

13. The method of claim 1, further comprising determining whether to send the bloom filter to the data storage system based on a costing function that at least considers a performance effect of applying the bloom filter at the data storage system.

14. The method of claim 1, wherein the filtered response is one or more filtered data blocks.

15. The method of claim 1, wherein the filtered response is comprises one of:

a row source, row set, or table.

16. The method of claim 1, wherein the filtered response comprises a virtual data block, comprising at least:

a payload comprised of a row set, and data block header or transmission information.

17. The method of claim 1,

wherein generating the bloom filter comprises populating an array in the bloom filter by applying one or more hash functions to attribute values of the one or more attributes of the second table;

wherein comparing results of the one or more functions of the values of the one or more data block row fields to the bloom filter comprises applying the same one or more hash functions to the values of the one or more data block row fields and determining whether the array has been populated at indexes to which the values of the one or more data block row fields are respectively hashed by the one or more hash functions.

18. A method comprising:

receiving, at a data storage system, an input/output (I/O) request for data from a database server, the I/O request specifying locations of a plurality of data blocks at which the requested data is stored;

wherein the data storage system comprises the one or more storage devices, upon which the database server stores data for table structures in a plurality of block structures, including the plurality of data blocks;

receiving, at the data storage system, in association with the I/O request, a bloom filter;

receiving, at the data storage system, metadata indicating to the data storage system one or more data block row fields, of the plurality of data blocks, that correspond to one or more particular columns against which the bloom filter is to be tested;

responsive to the I/O request;

the data storage system reading the plurality of data blocks from the one or more storage devices;

the data storage system generating filtered response data from the plurality of data blocks by removing data block rows that do not apply to the bloom filter, wherein removing data block rows that do not apply to the bloom filter comprises comparing results of one or more functions of the values of the one or more data block row fields to the bloom filter;

the data storage system sending the filtered response data to the database server;

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

19. The method of claim 18, wherein the instructions, when executed by the one or more computing devices, further cause:

the data storage system invoking a database library component to interpret the plurality of data blocks as table structures, the database library component comprising a subset of logic used by the database server to interpret data blocks as table structures.

20. The method of claim 18,

wherein comparing results of the one or more functions of the values of the one or more data block row fields to the bloom filter comprises applying one or more hash functions to the values of the one or more data block row fields and determining whether the array has been populated at indexes to which the values of the one or more data block row fields are respectively hashed by the one or more hash functions.

21. The method of claim 18, wherein the filtered response data is one or more filtered data blocks.

22. The method of claim 18, wherein the filtered response data comprises one of:

a row source, row set, or table.

23. The method of claim 18, wherein the filtered response data comprises a virtual data block, comprising at least:

a payload comprised of a row set, and data block header or transmission information.

24. The method of claim 18, further comprising:

receiving multiple I/O requests for different sets of one or more data blocks in which the data storage system stores data for the first table, the multiple I/O requests including the I/O request;

wherein at least some of the multiple I/O requests reference a copy of the bloom filter that is already stored at a location accessible to the data storage system.

25. The method of claim 18, further comprising:

receiving the bloom filter in the I/O request.

26. One or more non-transitory computer-readable media storing instructions which, when executed by the one or more computing devices, cause:

in a database system comprising at least a database server and a data storage system that is separate from the database server, performing a join operation between at least a first table and a second table by;

at the database server, generating a bloom filter based upon one or more attributes of the second table;

identifying, at the database server, one or more locations of one or more data blocks in which the data storage system stores data for the first table;

wherein the data storage system comprises one or more storage devices upon which data for tables, including the first table, is stored in a plurality of block structures, including the one or more data blocks;

the database server sending to the data storage system;

a) an input/output (I/O) request identifying the one or more locations of the one or more data blocks stored in the data storage system; and

b) the bloom filter, associated with the I/O request;

wherein the I/O request, is a communication that, when interpreted by the data storage system, causes;

the data storage system reading said one or more data blocks from the one or more storage devices;

the data storage system generating a filtered response to the I/O request from the one or more data blocks by removing data block rows that do not apply to the bloom filter; and

sending to the data storage system metadata indicating to the data storage system one or more data block row fields, of the one or more data blocks, that correspond to one or more particular columns of the first table, wherein removing data block rows that do not apply to the bloom filter comprises comparing results of one or more functions of the values of the one or more data block row fields to the bloom filter;

in response to the I/O request, the database server receiving the filtered response from the data storage system;

performing the join operation based on the filtered response.

27. The one or more storage media of claim 26, wherein the instructions, when executed by the one or more computing devices, further cause:

prior to sending the bloom filter to the data storage system, the database server receiving a second set of one or more data blocks from the data storage system;

the database server translating the one or more data blocks into the second table;

the database server generating a hash table based upon one or more attributes of a second table;

generating the bloom filter based upon the hash table.

28. The one or more non-transitory computer-readable media of claim 27, wherein the instructions, when executed by the one or more computing devices, further cause:

prior to sending the bloom filter to the data storage system, generating at least one predicate based upon the bloom filter;

sending the bloom filter to the data storage system in the form of the at least one predicate.

29. The one or more non-transitory computer-readable media of claim 26, wherein the filtered data includes one or more raw data blocks.

30. The one or more non-transitory computer-readable media of claim 26, wherein the I/O request identifies the one or more locations of the one or more data blocks by specifying one or more logical or physical addresses at which the one or more data blocks are stored.

31. The one or more non-transitory computer-readable media of claim 26, wherein the instructions, when executed by the one or more computing devices, further cause:

the database server translating the filtered response into only a partial version of the first table;

the database server performing the join operation based on the partial version of the first table.

32. The one or more non-transitory computer-readable storage-media of claim 26, wherein the bloom filter is a first bloom filter, the method further comprising:

at the database server, generating a second bloom filter based upon one or more attributes of a third table that is also involved in the join operation;

the database server sending to the data storage system;

(c) the second bloom filter, also associated with the I/O request;

wherein the I/O request further causes the data storage system generating the filtered response to the I/O request from the one or more data blocks by removing data block rows that do not apply to the second bloom filter.

33. The one or more non-transitory computer-readable storage-media of claim 26, wherein the instructions, when executed by the one or more computing devices, further cause sending the bloom filter in the I/O request.

34. The one or more non-transitory computer-readable storage-media of claim 26, wherein the instructions, when executed by the one or more computing devices, further cause:

sending multiple I/O requests to the data storage system for different sets of one or more data blocks in which the data storage system stores data for the first table, the multiple I/O requests including the I/O request;

wherein at least some of the multiple I/O requests reference a cached copy of the bloom filter stored at the data storage system responsive to the data storage system receiving the bloom filter from the database server.

35. The one or more non-transitory computer-readable media of claim 34, wherein the at least some of the multiple I/O requests are from database servers other than the database server.

36. The one or more non-transitory computer-readable media of claim 26, wherein the instructions, when executed by the one or more computing devices, further cause:

sending multiple I/O requests to a plurality of data storage systems, including the data storage system, for different sets of one or more data blocks in which the plurality of data storage systems store data for the first table, the multiple I/O requests including the I/O request;

sending the bloom filter to the data storage system by pushing the bloom filter to a location accessible to all of the plurality of storage systems, and referencing the bloom filter in the I/O request.

37. The one or more non-transitory computer-readable media of claim 26, wherein the instructions, when executed by the one or more computing devices, further cause:

prior to the I/O request, sending the bloom filter to the data storage system in association with a context identifier;

wherein the I/O request includes the context identifier to associate the I/O request with the bloom filter.

38. The one or more non-transitory computer-readable media of claim 26, wherein the instructions, when executed by the one or more computing devices, further cause determining whether to send the bloom filter to the data storage system based on a costing function that at least considers a performance effect of applying the bloom filter at the data storage system.

39. The one or more non-transitory computer-readable media of claim 26, wherein the filtered response is one or more filtered data blocks.

40. The one or more non-transitory computer-readable media of claim 26, wherein the filtered response is one of:

a row source, row set, table, or table.

41. The one or more non-transitory computer-readable media of claim 26, wherein the filtered response is a virtual data block comprising at least:

a payload comprised of a row set, and data block header or transmission information.

42. The one or more non-transitory computer-readable media of claim 26, wherein the instructions, when executed by the one or more computing devices, further cause the database server sending to the data storage system information indicating how the database server interprets the one or more data blocks.

43. The one or more non-transitory computer-readable media of claim 26,

wherein generating the bloom filter comprises populating an array in the bloom filter by applying one or more hash functions to attribute values of the one or more attributes of the second table;

wherein comparing results of the one or more functions of the values of the one or more data block row fields to the bloom filter comprises applying the same one or more hash functions to the values of the one or more data block row fields and determining whether the array has been populated at indexes to which the values of the one or more data block row fields are respectively hashed by the one or more hash functions.

44. One or more non-transitory storage-computer-readable media storing instructions which, when executed by one or more computing devices, cause performance of:

receiving, at a data storage system, an input/output (I/O) request for data from a database server, the I/O request specifying locations of a plurality of data blocks at which the requested data is stored;

wherein the data storage system comprises the one or more storage devices, upon which the database server stores data for table structures in a plurality of block structures, including the data blocks;

receiving, at the data storage system, in association with the I/O request, a bloom filter;

receiving, at the data storage system, metadata indicating to the data storage system one or more data block row fields, of the plurality of data blocks, that correspond to one or more particular columns against which the bloom filter is to be tested;

responsive to the I/O request;

the data storage system reading the plurality of data blocks from the one or more storage devices;

the data storage system generating filtered response data from the plurality of data blocks by removing data block rows that do not apply to the bloom filter, wherein removing data block rows that do not apply to the bloom filter comprises comparing results of one or more functions of the values of the one or more data block row fields to the bloom filter;

the data storage system sending the filtered response data to the database server.

45. The one or more storage media of claim 44, wherein the instructions, when executed by the one or more computing devices, further cause:

the data storage system invoking a database library component to interpret the plurality of data blocks as table structures, the database library component comprising a subset of logic used by the database server to interpret data blocks as table structures.

46. The one or more non-transitory computer-readable of claim 44,

wherein comparing results of the one or more functions of the values of the one or more data block row fields to the bloom filter comprises applying one or more hash functions to the values of the one or more data block row fields and determining whether the array has been populated at indexes to which the values of the one or more data block row fields are respectively hashed by the one or more hash functions.

47. The one or more non-transitory computer-readable media of claim 44, wherein the filtered response data is one or more filtered data blocks.

48. The one or more non-transitory computer-readable media of claim 44, wherein the filtered response data comprises one of:

a row source, row set, or table.

49. The one or more non-transitory computer-readable media of claim 44, wherein the filtered response data comprises a virtual data block, comprising at least:

a payload comprised of a row set, and data block header or transmission information.

50. The one or more non-transitory computer-readable media of claim 44, wherein the instructions, when executed by the one or more computing devices, further cause:

receiving multiple I/O requests for different sets of one or more data blocks in which the data storage system stores data for the first table, the multiple I/O requests including the I/O request;

wherein at least some of the multiple I/O requests reference a copy of the bloom filter that is already stored at a location accessible to the data storage system.

51. The one or more non-transitory computer-readable media of claim 44, wherein the instructions, when executed by the one or more computing devices, further cause:

receiving the bloom filter in the I/O request.

52. A system comprising:

one or more blocked storage devices storing data in a plurality of data block structures;

a storage server that responds to I/O requests for data block structures by reading data blocks from or writing data blocks to the one or more blocked storage devices; and

a database server, connected via a network to the storage server, that receives and responds to a given SQL query from a client by at least;

     1) requesting a first set of one or more data blocks from the storage server;

     2) interpreting the first set of one or more data blocks as one or more columns of at least one or more database tables; and

     3) executing a database operation indicated by the given SQL query against the at least one or more columns to produce a result set;

wherein, for a particular SQL query indicating a join operation between a first table and a second table, the database server is configured to generate a bloom filter based upon the second table and send to the storage server;

a) an I/O request that specifies one or more addresses from which to retrieve a second set of one or more data blocks storing data for the first table;

b) the bloom filter; and

c) metadata indicating to the data storage system one or more data block row fields of the one or more data blocks that correspond to one or more particular columns of the first table;

wherein the storage server is configured to, in response to the I/O request, read the second set of one or more data blocks from the one or more storage devices at the specified one or more addresses;

wherein the storage server is further configured to, upon reading the second set of one or more data blocks in response to the I/O request, produce filtered data by removing data block rows that do not apply to the bloom filter, wherein removing data block rows that do not apply to the bloom filter comprises comparing results of one or more functions of the values of the one or more data block row fields to the bloom filter;

wherein the storage server is further configured to, in response to the I/O request, send the filtered data to the database server; and

wherein the database server is further configured to perform the join operation based on the second table and the filtered data returned by the storage server.

53. An apparatus comprising:

one or more computer processors;

one or more network interfaces configured to send I/O requests to at least a data storage system, and receive responses to the I/O requests;

a database server component, implemented at least partially by the one or more computer processors, configured to perform database operations, including join operations between tables, wherein the database server component is configured to perform a particular join operation between at least a first table and a second table by;

generating a bloom filter based upon one or more attributes of the second table;

identifying one or more locations of one or more data blocks in which the data storage system stores data for the first table;

wherein the data storage system comprises one or more storage devices upon which data for tables, including the first table, is stored in a plurality of block structures, including the one or more data blocks;

sending to the data storage system, over the one or more network interfaces;

a) an input/output (I/O) request identifying the one or more locations of the one or more data blocks stored in the data storage system; and

b) the bloom filter, associated with the I/O request;

wherein the I/O request, is a communication that, when interpreted by the data storage system, causes;

the data storage system reading said one or more data blocks from the one or more storage devices;

the data storage system generating a filtered response to the I/O request from the one or more data blocks by removing data block rows that do not apply to the bloom filter; and

sending to the data storage system, over the one or more network interfaces, metadata indicating to the data storage system one or more data block row fields, of the one or more data blocks, that correspond to one or more particular columns of the first table, wherein removing data block rows that do not apply to the bloom filter comprises comparing results of one or more functions of the values of the one or more data block row fields to the bloom filter;

in response to the I/O request, receiving the filtered response from the data storage system over the one or more network interfaces; and

performing the particular join operation based on the filtered response.

54. An apparatus comprising:

one or more storage devices, upon which is stored a plurality of block structures;

one or more network interfaces configured to receive and send responses to I/O requests from one or more database servers that store data for table structures within the plurality of block structures;

a database storage server component, implemented at least partially by hardware, configured to process the I/O requests, wherein the data storage server component is configured to process a particular I/O request by;

receiving the particular I/O request from a particular database server, the particular I/O request specifying particular locations of a plurality of data blocks;

receiving, in association with the particular I/O request, a bloom filter;

receiving metadata indicating one or more data block row fields, of the plurality of data blocks, that correspond to one or more particular columns against which the bloom filter is to be tested;

reading the plurality of data blocks from the one or more storage devices;

generating filtered response data from the plurality of data blocks by removing data block rows that do not apply to the bloom filter, wherein removing data block rows that do not apply to the bloom filter comprises comparing results of one or more functions of the values of the one or more data block row fields to the bloom filter; and

sending the filtered response data to the particular database server.