Method for external sorting in shared-nothing parallel architectures

A distributed relational database system for parallel sorting of a relation which is a set of tuples to be sorted on multiple sites, the system comprising:

• a) an interconnection network;
• b) one or more logical sites, each site having a least one processor with memory thereon in communication with the interconnection network, the logical sites including i) one or more storage sites, ii) two or more sort sites, each sort site having at least a predetermined memory capacity, iii) one or more sink sites, and iv) at least one coordinator site;
• and
• c) an executable program for performing a parallel sort, comprising the steps of:
• determining global interval key values such that interval fits in the predetermined memory capacity of a single sort site, notwithstanding that the number of global interval key values might exceed the number of sort sites, wherein the tuples between two global key values define an interval,
• scanning the relation distributed to each storage site,
• determining for each tuple an assigned interval and a corresponding assigned sort site, and
• sending each tuple to the assigned sort site.

A sort node, for use in a distributed relational database, the distributed relational database having a plurality of nodes configured in a network, the network including a plurality of sort sites and a sink site, each of the sort sites corresponding with a respective interval, and each sort site having at least a predetermined memory capacity, the respective intervals being defined as falling between adjacent ones of a sequence of interval key values, the sort node comprising:

• a) processor means for processing executable steps;
• b) memory means, associated with said processor means, for storing data;
• c) storage means for storing database information;
• d) communication hardware means for enabling said sort node to communicate across said network;
• e) means for receiving a set of global interval key values and intervals therebetween, such that a predetermined one of the intervals corresponds with said sort node and fits within the predetermined memory capacity of a respective one of the sort sites, notwithstanding that the number of a global interval key values might exceed the number of sort sites;
• f) means for storing tuples in temporary files at said sort node;
• and
• g) means for repeating, for each interval on each of the sort sites, the executable steps of:
• 1) reading an interval;
• 2) performing an in-memory sort of the tuples stored, for the internal read;
• and
• 3) sending the tuples of the sorted interval to the sink site.

A storage node, for use in a distributed relational database, the distributed relational database having a plurality of nodes configured in a network, the network including a designated coordinator site and a plurality of sort sites, each sort site having at least a predetermined memory capacity, the storage node comprising:

• a) processor means for processing executable steps;
• b) memory means, associated with said processor means, for storing data;
• c) storage means for storing database information;
• d) communication hardware means for enabling said storage node to communicate across said network,
• e) means for generating and sorting a local sample at said storage node;
• f) means for sending the local random sample from said storage node to the designated coordinator site;
• g) means for receiving a relation, the relation including a sequence of interval key values, intervals being defined between adjacent ones of the sequence of interval key values, such that each interval fits in the predetermined memory capacity of a single sort site, notwithstanding that the number of global interval key values might exceed the number of sort sites, each of the intervals corresponding with a respective one of the sort sites;
• h) means for scanning the relation and for determining, for each of a set of tuples to be sorted, an interval within which the tuple falls, and the corresponding sort site;
• and
• i) means for sending each tuple to another site across said network.

A coordinator node, for use in a distributed relational database, the distributed relational database having a plurality of nodes configured in a network, the network including a plurality of other sites, the plurality of other sites including a plurality of sort sites, each of the sort sites having a main memory including at least a predetermined memory capacity, the coordinator node comprising:

• a) processor means for processing executable steps;
• b) memory means, associated with said processor means, for storing data;
• c) storage means for storing database information;
• d) communication hardware means for enabling said coordinator node to communicate across said network;
• e) means for merging local random samples on said coordinator node to provide a single global sample;
• f) means for determining a sequence of global key values such that the tuples between two adjacent ones of the sequence of global key values define a respective interval, and each respective interval fits in the predetermined memory capacity of a respective one of the sort sites, notwithstanding that the number of global interval key values might exceed the number of sort sites, and
• g) means for sending the global interval key values to the other sites across said network.