Scheduled network communication for efficient re-partitioning of data

Patent metadata

Field	Value
Publication number	US-9813490-B2
Application number	US-201514711617-A
Country	US
Kind code	B2
Filing date	May 13, 2015
Priority date	May 13, 2015
Publication date	Nov 7, 2017
Grant date	Nov 7, 2017

Abstract

A method, apparatus, and system for efficiently re-partitioning data using scheduled network communication are provided. Given re-partitioning data defining the data blocks to be sent amongst a plurality of server nodes, a corresponding network schedule is determined to send the data blocks in a coordinated manner. The network schedule is divided into time slots, wherein each of the plurality of server nodes can send up to one data block and receive up to one data block in each time slot. By using a greedy selection algorithm that prioritizes by largest senders and largest receivers, a near optimal schedule can be determined even in the presence of heavy skew. The greedy selection algorithm can be implemented with a O(T*N^2) time complexity, enabling scaling to large multi-node clusters with many server nodes. The network schedule is of particular interest for database execution plans requiring re-partitioning on operators with different keys.

First claim

What is claimed is:

1. A method comprising:
receiving re-partitioning data describing, for each server node of a plurality of server nodes connected by a network, a quantity of data blocks to be sent to said each server node;
determining a sender order for the plurality of server nodes by using the re-partitioning data to sort, in descending order, the plurality of server nodes by a total quantity of data blocks to be sent;
according to the sender order, populating a network schedule comprising a plurality of time slots, wherein populating a network schedule includes for each time slot of the plurality of time slots;
assigning in the sender order at most one receiver node, of the plurality of server nodes to receive one or more data blocks over the network from a sender node of said plurality of server nodes, wherein assigning at most one receiver node is based at least on having a largest possible quantity of data blocks to be received by the sender node from any server node of said plurality of server nodes not yet assigned as a receiver node in said each time slot;
wherein each server node of said plurality of server nodes is assigned as a receiver node no more than once in said each time slot;

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

2. The method of claim 1 , wherein for a particular time slot of said plurality of time slots and particular server node of said plurality of server nodes, assigning in the sender order at most one receiver node includes traversing an ordered list that includes other server nodes of said plurality of server nodes, wherein the ordered list is ordered in descending order of a quantity of data blocks to be received by the particular server node from each of the other server nodes of said plurality of server nodes.

3. The method of claim 2 , wherein the traversing uses a bitmask that is maintained for said particular time slot to ensure that for said particular time slot a particular node of said plurality of server nodes is assigned as a receiver node no more than once in said particular time slot.

4. The method of claim 1 , wherein for a particular time slot of said plurality of time slots, assigning in the sender order at most one receiver node is based on populating contiguous time slots with the at most one receiver node.

5. The method of claim 1 , wherein the re-partitioning data is for transitioning from a first database operator on a first key to a second database operator on a second key.

6. The method of claim 1 , wherein the one or more computing devices comprises a coordinator server.

7. The method of claim 1 , wherein a size of each of the quantity of data blocks and a length of each of the plurality of time slots are based on the network.

8. A non-transitory computer-readable medium storing one or more sequences of instructions which, when executed by one or more processors, cause:
receiving re-partitioning data describing, for each server node of a plurality of server nodesconnected by a network, a quantity of data blocks to be sent to said each server node;
determining a sender order for the plurality of server nodes by using the re-partitioning data to sort, in descending order, the plurality of server nodes by a total quantity of data blocks to be sent;
according to the sender order, populating a network schedule comprising a plurality of time slots, wherein populating a network schedule includes for each time slot of the plurality of time slots,
assigning in the sender order at most one receiver node of the plurality of server nodes to receive one or more data blocks over the network from a sender node of said plurality of server nodes, wherein assigning at most one receiver node is based at least on having a largest possible quantity of data blocks to be received by the sender nodefrom any server node of said plurality of server nodes not yet assigned as a receiver node in said each time slot; and
wherein each server node of said plurality of server nodes is assigned as a receiver node no more than once in said each time slot.

9. The non-transitory computer-readable medium of claim 8 , wherein for a particular time slot of said plurality of time slots and particular server node of said plurality of server nodes, assigning in the sender order at most one receiver node includes traversing an ordered list that includes other server nodes of said plurality of server nodes, wherein the ordered list is ordered in descending order of a quantity of data blocks to be received by the particular server node from each of the other server nodes of said plurality of server nodes.

10. The non-transitory computer-readable medium of claim 9 , wherein the traversing uses a bitmask that is maintained for said particular time slot to ensure that for said particular time slot a particular node of said plurality of server nodes is assigned as a receiver node no more than once in said particular time slot.

11. The non-transitory computer-readable medium of claim 8 , wherein for a particular time slot of said plurality of time slots, assigning in the sender order at most one receiver node is based on populating contiguous time slots with the at most one receiver node.

12. The non-transitory computer-readable medium of claim 8 , wherein the re-partitioning data is for transitioning from a first database operator on a first key to a second database operator on a second key.

13. The non-transitory computer-readable medium of claim 8 , wherein a size of each of the quantity of data blocks and a length of each of the plurality of time slots are based on the network.

14. A system comprising one or more computing devices configured to:
receive re-partitioning data describing, for each server node of a plurality of server nodes connected by a network, a quantity of data blocks to be sent to said each server node;
determine a sender order for the plurality of server nodes by using the re-partitioning data to sort, in descending order, the plurality of server nodes by a total quantity of data blocks to be sent;
according to the sender order, to populate a network schedule comprising a plurality of time slots, wherein to populate a network schedule said system is confugured to, for each time slot of the plurality of time slots:
to assign in the sender order at most one receiver node of the plurality of server nodes to receive one or more data blocks over the network from a sender node of said plurality of server nodes, wherein to assign at most one receiver node is based at least on having a largest possible quantity of data blocks to be received by the sender nodefrom any server node of said plurality of server nodes not yet assigned as a receiver node in said each time slot; and

wherein each server node of said plurality of server nodes is assigned as a receiver node no more than once in said each time slot.

15. The system of claim 14 , said system is configured for a particular time slot of said plurality of time slots and particular server node of said plurality of server nodes, to assign in the sender order at most one receiver node by traversing an ordered list that includes other server nodes of said plurality of server nodes, wherein the ordered list being ordered in descending order of a quantity of data Hocks to be received by the particular server node from each of the other server nodes of said plurality of server nodes
for a particular time slot of said plurality of time slots, to assign in the sender order at most one receiver node by traversing an ordered list of said plurality of server nodes, wherein the ordered list is ordered in descending order

Assignees

• Oracle Int Corp

Inventors

• Idicula Sam
• Basant Aarti
• Aggarwal Vikas
• Wolf Stephan
• Agarwal Nipun

Classifications

• G06F16/2471

  Distributed queries · CPC title

• G06F16/278

  Data partitioning, e.g. horizontal or vertical partitioning · CPC title

• H04L12/40143 Primary

  involving priority mechanisms (hybrid switching fabrics H04L12/6402; intermediate storage or scheduling H04L49/90; time-division multiplex systems H04J3/00) · CPC title

• H04L47/828

  Allocation of resources per group of connections, e.g. per group of users · CPC title

• H04L67/10 Primary

  in which an application is distributed across nodes in the network (software deployment G06F8/60; multiprogramming arrangements G06F9/46) · CPC title