A Guide to Sharding-Sphere Series — Result Attribution

10 min readSep 29, 2018

“A Guide to Sharding-Sphere” is a series of articles written by core developers of Sharding-Sphere to introduce and analyse core models, advanced technologies adopted and valuable experience summarised, etc. The series will lead you to the core of Sharding-Sphere and make you acquire new knowledge and inspirations. Looking forward to your focuses, we hope to communicate and develop along with you!

Introduction to the Lecturer:

Passionate to open source, Zhang Liang is the former director of Architecture Department, Dangdang, who is in charge of two open-source projects currently, i.e., Elastic-Job and Sharding-Sphere (Sharding-JDBC). Advocating elegant coding, he is adept at java dominated distributed architecting and Kubernetes and Mesos dominated cloud platform, with much research on how to produce expressive codes. Joining JD.com at the beginning of 2018, he is responsible for data research and devoted to building Sharding-Sphere as the first-rate data solution in the finance industry now.

Introduction

Result merger is to merge multi-data result set acquired from all the data nodes as one result set and return it to the request end rightly.

The result merger supported by Sharding-Sphere can be divided into 4 kinds in function (iteration, order-by, group-by and pagination) and 3 kinds in implementation (stream, memory and decorator). The former division is in the relationship of composition rather than clash, while the later one is clashing in its streaming merger and the memory merger, and the decorator merger can be further processed based on the streaming merger and the memory merger.

Able to reduce the memory consumption to a large extend, the most prior choice of merger method is to follow the database returned result set, since the result set is returned from database line by line, instead of loading all the data to the memory at one time.

Streaming merger means, each time, the data acquired from the result set is able to return the single piece of right data line by line. It is the most suitable one for the original result set return method of the database. Iteration, order-by, and stream group-by all belong to the range of streaming merger.

Memory merger requires to iterate all the data in the result set and store it in the memory, and pack it into data result set visited line by line to return, after unified computations of grouping, ordering and aggregation.

Decorator merger merges and reinforce all the result sets function uniformly. Currently, decorator merger has only one type, pagination merger.

Iteration Merger

As the simplest merger method, iteration merger requires only the combination of multiple data result sets into a singly linked list. After iterating current data result sets in the linked list, it only needs to move the element of linked list to the next position and iterate the next data result set.

Order-by Merger

For the existence of ORDER BY sentence in SQL, each data result has its own order, so ordering the data value currently pointed by cursors in the result set is enough. It is equal to sequencing multiple ordered arrays, and therefore, order-by merger is the most suitable ordering algorithm in this situation.

When merging order inquiries, Sharding-Sphere will compare current data values of each result set, which is realized by Comparable interface of Java, and put them into the priority queue. Each time when acquiring the next piece of data, it only needs to move the result set in the top end of the line down, renter the priority order according to the new cursor and relocate its own position.

Here is to explain the order-by merger of Sharding-Sphere. The following is an example diagram of ordering by the score.

Here are data result sets returned by 3 diagrams shown in the example, and each one of them has already been ordered according to the score, but there is no order among the 3 data result sets. When ordering the data value currently pointed by cursors in these 3 result sets and putting them into the priority queue, the data value of t_score_0 is the biggest, followed by that of t_score_2 and t_score_1 in sequence. Thus, the priority queue is ordered by the order of t_score_0, t_score_2 and t_score_1.

This diagram illustrates how the order-by merger works when using next invocation.

We can see from the diagram that when using next invocation, t_score_0 at the first of the queue will be popped out. After returning the data value currently pointed by the cursor, i.e., 100, to the client end, the cursor will be moved down and t_score_0 will be put back to the queue. While the priority queue will also be ordered according to the t_score_0 data value (90 here) pointed by the cursor of current data result set. According to the current value, t_score_0 is at the last of the queue, and in the second place of the queue formerly, the data result set of t_score_2, automatically moves to the first place of the queue.

In the second use of next, t_score_2 in the first position is popped out of the queue. Its value pointed by the cursor of the data result set is returned to the client end, with its cursor moved down to rejoin the queue, and the following will be in the same way. If there is no data in the result set, it will not rejoin the queue.

It can be seen that, under the circumstance that single data result set is ordered while multiple data result set is disordered, Sharding-Sphere do not need to upload all the data to the memory to order. By the order-by merger method, next only acquires one right piece of data each time, so the memory consumption can be saved to a large extent.

On the other hand, the order-by merger of Sharding-Sphere has its order on horizontal axis and vertical axis that maintains the data result set. Naturally ordered, vertical axis refers to each data result set itself, which is acquired by SQL that contains ORDER BY. Horizontal axis refers to the current value pointed by each data result set, and its order needs to be maintained by the priority queue. Each moving down of the current cursor of data result set requires to put it in the priority order again, which means the operation of cursor moving down can only happen when the data result set is in the first place of the queue.

Group-by Merger

With the most complicated situation, group-by merger can be divided into stream group-by merger and memory group-by merger. Stream group-by merger requires the field and order type (ASC or DESC) of SQL order item and group-by item to be consistent. Otherwise, its data accuracy can only be maintained by memory merger.

For instance, suppose the sheet structure is divided according to the subject, and each sheet contains each examinee’s name (to simplify, the same name is not taken into consideration) and score. The SQL used to acquire each examinee’s total score is as follow:

SELECT name, SUM(score) FROM t_score GROUP BY name ORDER BY name;

When order-by item and group-by item are totally consistent, the data obtained are continuous. For the data needed to group are all stored in the data value currently pointed by cursors of each data result set, stream group-by merger can be used, as illustrated by the diagram:

The merging logic will be similar as that of order-by merger. The following picture shows how stream group-by merger works when using next invocation.

We can see from the picture, in the first next invocation, t_score_java in the first position, along with the data having the grouping value of “Jetty” in other result sets, will be popped out of the queue. After acquiring all the students’ scores with the name of “Jetty”, the accumulation operation will be proceeded. Hence, after the first next invocation is finished, the result set acquired is the sum of Jetty’s scores. In the same time, all the cursors in data result sets will be moved down to the next different data value of “Jetty”, and rearranged according to the value currently pointed by cursors of the result set. Thus, the data that contains the second-place name “John” will be put at the beginning of the queue

Stream group-by merger is different from order-by merger only in two points:

1. It will take out all the data with the same grouping item from multiple data result sets once.

2. It does the aggregation calculation according to the type of aggregation calculation function.

For the dis-conformation between the grouping item and the order-by item, it requires to upload all the result set data to the memory to group and aggregate, since the relevant data value needed to acquire grouping information is not continuous, and stream merger is not able to use. For example, if acquiring each examinee’s total score through the following SQL and ordering them from the highest to the lowest:

SELECT name, SUM(score) FROM t_score GROUP BY name ORDER BY score DESC;

Then, stream merger is not able to use, for the data taken out from each data result set is the same as the original data of the diagram ordered by score in the upper half part structure.

When there is only SQL contained in the group-by sentence, according to the realization of different databases, its order may not be the same as the grouping order. The lack of ordering sentence indicates that the order is not emphasized in this SQL. Therefore, through the adaption of SQL optimization, Sharding-Sphere can automatically add the ordering item consistent with the grouping item, converting it from the memory group-by merger method that consumes the memory to the stream group-by merger scheme.

Whether stream group-by merger or memory group-by merger, processes in the identical way of the aggregation function, which can be categorized into 3 kinds: comparison, sum and average.

Comparison aggregation function refers to MAX and MIN. They need to compare all the result set data and return its maximum or minimum value directly.

Sum aggregation function refers to SUM and COUNT. They need to sum up all the result set data.

Average aggregation function refers only to AVG. it must be calculated through the adaption of SQL optimization, which has been covered above in the content of SQL adaption, so we will state no more here.

Pagination Merger

All the merger types stated above can be paginated. Pagination is the decorator added on other kinds of mergers, through the decorator model of which, Sharding-Sphere augments its ability to paginate the data result set. Pagination merger is responsible for filtering the data unnecessary to acquire.

The pagination function of Sharding-Sphere can be misleading to users in that they may think it will take a large amount of memory. In distributed situations, it can only guarantee the data accuracy by readapting LIMIT 10000000, 10 to LIMIT 0, 10000010. Users can easily bear the misconception that Sharding-Sphere uploads a large amount of meaningless data to the memory and causes the risk of memory overflow. Actually, it can be known from the principle of stream merger, only the case of memory group-by merger will upload all the data to the memory. Generally speaking, however, SQL used for OLAP grouping, is applied more frequently to massive calculation or small result generation situation, rather than generates vast result data. Except for the case of memory group-by merger, stream merger is the method to acquire data result set in other cases: Sharding-Sphere would skip the data unnecessary to take by the method of next in result set, rather than storing them in the memory.

What’s to be noticed, pagination with LIMIT is not the best practice actually, because a large amount of data still needs to be transmitted to the memory space of Sharding-Sphere for the sake of ordering. LIMIT cannot search for data by index, so paginating with ID is a better solution on the premise that the ID continuity can be guaranteed. For example:

SELECT * FROM t_order WHERE id > 100000 AND id <= 100010 ORDER BY id;

Or search the next page through the ID of the last query result record, for example:

SELECT * FROM t_order WHERE id > 10000000 LIMIT 10;