Effective and scalable building and probing of hash tables using multiple GPUs
Described approaches provide for effectively and scalably using multiple GPUs to build and probe hash tables and materialize results of probes. Random memory accesses by the GPUs to build and/or probe a hash table may be distributed across GPUs and executed concurrently using global location identifiers. A global location identifier may be computed from data of an entry and identify a global location for an insertion and/or probe using the entry. The global location identifier may be used by a GPU to determine whether to perform an insertion or probe using an entry and/or where the insertion or probe is to be performed. To coordinate GPUs in materializing results of probing a hash table a global offset to the global output buffer may be maintained in memory accessible to each of the GPUs or the GPUs may compute global offsets using an exclusive sum of the local output buffer sizes.
1. A method comprising:
receiving one or more queries indicating one or more join operations corresponding to one or more tables stored in one or more relational databases;
generating a hash table comprising hash values corresponding to one or more first portions of the one or more tables, wherein a plurality of graphics processing units (GPUs) are assigned respective partitions of the hash table that represent respective parts of the hash table;
computing, from an entry corresponding to one or more second portions of the one or more tables, a global location identifier that corresponds to one or more locations in the hash table;
determining, using the global location identifier, the entry corresponds to at least one hash table partition of the partitions of the hash table, the at least one hash table partition assigned to a GPU of the plurality of GPUs; and
probing, using the GPU and based at least on the entry being determined to correspond to the at least one hash table partition assigned to the GPU, the at least one hash table partition corresponding to the GPU to produce one or more results of the one or more join operations.
2. The method of claim 1 , wherein the probing is performed based at least on determining that a hash value of one or more keys of the entry falls within the at least one hash table partition assigned to the GPU.
3. The method of claim 1 , wherein the global location identifier includes a global hash index that identifies a location for the entry in the hash table.
4. The method of claim 1 , further comprising:
filtering, using the GPU, entries from at least a portion of a probe table corresponding to the one or more second portions of the one or more tables based at least on the global location identifier indicating the entries correspond to the GPU to generate a filtered portion of the probe table that does not include the entries; and
transmitting the filtered portion of the probe table to a different GPU of the plurality of GPUs, the transmitting causing one or more probes of the hash table using of one or more entries in the filtered portion of the probe table.
5. The method of claim 1 , wherein the entry is received, by the GPU, from a different GPU of the plurality of GPUs based at least on the entry being determined to correspond to the at least one hash table partition assigned to the GPU.
6. The method of claim 1 , wherein:
The GPU reads the entry from a first portion of a probe table corresponding to the one or more second portions of the one or more tables in a first buffer of the GPU used to analyze the entry in an iteration of the GPU iteratively analyzing entries of the probe table for probing the hash table, and the method further includes:
receiving, from a different GPU of the plurality of GPUs, a second portion of the probe table at a second buffer of the GPU that is used as a staging buffer in the iteration for a subsequent iteration of the GPU iteratively analyzing entries of the probe table for probing the hash table.
7. The method of claim 1 , wherein the one or more queries include one or more OnLine Analytical Processing queries.
8. One or more processors comprising:
one or more circuits to:
receive one or more queries indicating one or more join operations corresponding to one or more tables stored in one or more relational databases;
read, using a first graphics processing unit (GPU) of a plurality of GPUs, an entry corresponding to one or more first portions of the one or more tables;
compute, from the entry, a global location identifier that corresponds to one or more locations in a hash table comprising hash values corresponding to one or more second portions of the one or more tables, wherein the plurality of GPUs are assigned respective partitions of the hash table that represent respective parts of the hash table;
determine, using the global location identifier, the entry corresponds to at least one hash table partition of the partitions of the hash table, the at least one hash table partition assigned to a second GPU of the plurality of GPUs; and
cause the at least one hash table partition to be probed to produce one or more results of the one or more join operations based at least on the entry being determined to correspond to the at least one hash table partition assigned to the second GPU.
9. The one or more processors of claim 8 , wherein the causing the at least one hash table partition to be probed includes transmitting, using the first GPU, at least the entry for probing of the at least one hash table partition using the second GPU.
10. The one or more processors of claim 8 , wherein the causing the hash table partition to be probed includes remotely probing, using the first GPU, the at least one hash table partition at the second GPU.
11. The one or more processors of claim 8 , wherein the causing the at least one hash table partition to be probed is based at least on determining that a hash value of one or more keys of the entry falls within the at least one hash table partition assigned to the second GPU.
12. The one or more processors of claim 8 , wherein the global location identifier includes a global hash index that identifies a location for the entry in the hash table.
13. The one or more processors of claim 8 , wherein the one or more circuits are further to:
filter, using the first GPU, entries from at least a portion of a probe table corresponding to the one or more first portions of the one or more tables based at least on the global location identifier indicating the entries correspond to the first GPU to generate a filtered portion of the probe table that does not include the entries; and
transmit the filtered portion of the probe table to a different GPU of the plurality of GPUs, the transmitting causing one or more probes of the hash table using of one or more entries included in the filtered portion of the probe table.
14. The one or more processors of claim 8 , wherein at least a portion of a probe table corresponding to the one or more first portions of the one or more tables is received, by the first GPU, from a different GPU of the plurality of GPUs based at least on the entry being determined to correspond to the at least one hash table partition assigned to the second GPU.
15. The one or more processors of claim 8 , wherein:
the reading the entry is from a first portion of a probe table corresponding to the one or more first portions of the one or more tables in a first buffer of the GPU used to analyze the entry in an iteration of the GPU iteratively analyzing entries of the probe table for probing the hash table, and the one or more circuits are further to:
receive, from a different GPU of the plurality of GPUs, a second portion of the probe table at a second buffer of the GPU that is used as a staging buffer in the iteration for a subsequent iteration of the GPU iteratively analyzing entries of the probe table for probing the hash table.
16. The one or more processors of claim 8 , wherein the one or more queries include one or more OnLine Analytical Processing queries.
17. A system comprising:
a plurality of graphics processing units (GPUs) to perform collaborative probing of a hash table corresponding to one or more first portions of one or more tables stored in one or more relational databases based at least on one or more queries indicating one or more operations corresponding to the one or more tables, wherein the plurality of GPUs are assigned respective partitions of the hash table that represent respective parts of the hash table, and
wherein a GPU of the plurality of GPUs:
determines, from data read from an entry corresponding to one or more second portions of the one or more tables, a global location identifier that corresponds to one or more locations in the hash table, and
determines, using the global location identifier, the entry corresponds to at least one hash table partition of the partitions of the hash table, the at least one hash table partition assigned to a designated GPU from the plurality of GPUs;
wherein the at least one hash table partition is probed, using the designated GPU, to produce one or more results of the one or more operations based at least on the entry being determined to correspond to the at least one hash table partition assigned to the designated GPU.
18. The system of claim 17 , wherein the GPU provides at least the entry to a different one of the plurality of GPUs based at least on the GPU determining, using the global location identifier, the designated GPU is different than the GPU.
19. The system of claim 17 , wherein the GPU probes the hash table partition based at least on the global location identifier indicating the GPU is the designated GPU.
20. The system of claim 17 , wherein the plurality of GPUs are logically arranged in a ring, and the probing further includes each GPU of the plurality of GPUs providing a plurality of entries of a respective portion of a probe table corresponding to the one or more second portions of the one or more tables to a neighbor GPU in the ring.