Aggregation operations in a distributed database

1. A method for query planning in a distributed database that includes a table partitioned into shards distributed to database instances of the distributed database, comprising:

receiving a data-query at a query coordinator,

wherein the data-query comprises a first “distinct count” clause on a first column of the table and a “group by” clause on least a second column of the table, and

wherein the table is partitioned into the shards according to a sharding criterion; formulating a query plan to include:

respective instructions for converting, at at least some of the database instances, distinct values of the first column grouped by values of the second column into a count of the distinct values grouped by the values of the second column to obtain respective intermediate results;

instructions for receiving, at the query coordinator, the respective intermediate results from at least a subset of the at least some of the database instances; and

instructions for concatenating the respective intermediate results using a summing operation to obtain the first “distinct count” of the first column grouped by the second column;

executing the query plan to obtain results data; and

outputting the results data.

2. The method of claim 1 , wherein the sharding criterion consists of the first column.

3. The method of claim 1 , wherein the sharding criterion comprises the first column and the second column.

4. The method of claim 1 ,

wherein the data-query comprises a second “distinct count” clause on a third column, and

wherein the sharding criterion does not include the third column.

5. The method of claim 4 , wherein the query plan further comprises:

respective instructions for including, at the at least some of the database instances, respective distinct values of the third column grouped by the first column; and

instructions for performing, for each distinct value of the first column, a union aggregation of all the respective distinct values of the third column grouped by the first column received from the database instances.

6. The method of claim 1 , wherein the query plan is such that distinct values of the first column are not transmitted from at least some of the database instances to the query coordinator.

7. The method of claim 1 , wherein executing the query plan to obtain results data comprises:

transmitting respective portions of the query plan to at least some of the database instances; and

receiving respective portions of the results data from the at least some of the database instances.

8. A system for query planning in a distributed database that includes a table partitioned into shards distributed to database instances of the distributed database, comprising

a memory; and

a processor, the processor configured to execute instructions stored in the memory to:

formulate a query plan for a data-query that includes a first “distinct count” clause on a first column of the table and a “group by” clause on least a second column of the table which is partitioned into the shards according to a sharding criterion, wherein to formulate the query plan comprises instructions to:

generate, for inclusion in the query plan, respective instructions for converting, at at least some of the database instances, distinct values of the first column grouped by values of the second column into a count of the distinct values grouped by the values of the second column to obtain respective intermediate results;

generate, for inclusion in the query plan, instructions for receiving the respective intermediate results from at least a subset of the at least some of the database instances; and

generate, for inclusion in the query plan, instructions for concatenating the respective intermediate results using a summing operation to obtain the first “distinct count” of the first column grouped by the second column;

execute the query plan to obtain results data; and

outputting the results data.

9. The system of claim 8 , wherein the sharding criterion consists of the first column.

10. The system of claim 8 , wherein the sharding criterion comprises the first column and the second column.

11. The system of claim 8 ,

wherein the data-query comprises a second “distinct count” clause on a third column, and

wherein the sharding criterion does not include the third column.

12. The system of claim 11 , wherein to formulate the query plan further includes instructions to:

generate, for inclusion in the query plan, respective instructions for including, at the at least some of the database instances, respective distinct values of the third column grouped by the first column; and

generate, for inclusion in the query plan, instructions for performing, for each distinct value of the first column, a union aggregation of all the respective distinct values of the third column grouped by the first column received from the database instances.

13. The system of claim 8 , wherein the query plan is such that distinct values of the first column are not transmitted from the database instances.

14. The system of claim 8 , wherein to execute the query plan to obtain the results data comprises to:

transmit respective portions of the query plan to at least some of the database instances; and

receive respective portions of the results data from the at least some of the database instances.

15. A non-transitory computer readable medium storing instructions operable to cause one or more processors to perform operations for query planning in a distributed database that includes a table partitioned into shards distributed to database instances of the distributed database and sharded according to a sharding criterion, the operations comprising:

receiving a data-query at a query coordinator to perform a first “distinct count” on a first column of the table and a “group by” on least a second column of the table;

formulating a query plan to include:

respective instructions for converting, at at least some of the database instances, distinct values of the first column grouped by values of the second column into a count of the distinct values grouped by the values of the second column to obtain respective intermediate results;

instructions for receiving, at the query coordinator, the respective intermediate results from at least a subset of the at least some of the database instances; and

instructions for concatenating the respective intermediate results using a summing operation to obtain the first “distinct count” of the first column grouped by the second column;

executing the query plan to obtain results data; and

outputting the results data.

16. The non-transitory computer readable medium of claim 15 , wherein the sharding criterion consists of the first column.

17. The non-transitory computer readable medium of claim 15 , wherein the sharding criterion comprises the first column and the second column.

18. The non-transitory computer readable medium of claim 15 ,

wherein the data-query comprises a second “distinct count” clause on a third column not included in the sharding criterion.

19. The non-transitory computer readable medium of claim 18 , wherein the query plan further comprises:

instructions for performing, for each distinct value of the first column, a union aggregation of all respective distinct values of the third column grouped by the first column received from the database instances.

20. The non-transitory computer readable medium of claim 15 , wherein the query plan is such that distinct values of the first column are not transmitted from at least some of the database instances to the query coordinator.