Background
Geode supports allows querying data from stored in regions using query language called OQL is used for querying data from Geode. Currently it doesn’t have support for aggregate functions, except supporting “count(*)” that is without the accompanying group by functionality.
Our customer have asked for this feature multiple times.
Requirements
. Currently the query engine supports COUNT aggregate function but missing other standard functions like MIN, MAX, AVG, SUM. This document is to propose implementing Aggregate functions (with GROUP BY extension) and also to support UDA(user defined aggregate) functionalities with a common a frame work.
Requirement
Enhance the OQL engine to support:
- Aggregate functions (AVG , MaxMAX, MinMIN, CountCOUNT, SumSUM)
- Associated Group By GROUP BY clause
- Investigation on UDA ( User Defined Aggregates) implementation
Proposal
To implement the functionality , broadly , following need to be done:
A common framework to support both Aggregate and UDAs
Proposal
Provide a common interface that will be used to support both standard (internal) and UDAs.
OQL engine will provide a common interface "com.gemstone.gemfire.cache.query.Aggregator" that will defined the methods used to support both standard aggregates and UDA.
The interface "Aggregator" is described in details in the later section of the document (UDA).
OQL Grammar changes :
- Support non distinct order by
- Detect Group By Clause
- Detect Group By Path expressions
- Detect Aggregate functions in the projection attribute
Query correctness criteria:
...
a) If the query contains Group By Clause, then the projection attribute should only contain aggregate functions & only those columns which are present in Group By Clause.eg
E.g.:
select col1, avg( col2) from /
...
portfolio group by col1 // Valid query
...
select col1, col3, avg( col2) from /
...
portfolio group by col1
...
// invalid query (col3 missing from group by clause)
b) If the query does not contain Group By clause then its projection can either not contain any aggregate functions or they can only contain aggregate functions.eg
E.g.:
select avg( col2) from /portfolio /
...
/ Valid
select col1, avg ( col2) from /portfolio
...
(col1 should not be present in projection attributes) // invalid
Query Types & implementation:
Implementation for queries containing only group by clause & projection containing only columns (no aggregate function)
Select pf.ID , pf.status as status from /portfolio pf where pf.ID > 100 group by pf.ID , status
The above query can be converted into a distinct – order by query to get the result.
Transformed query => select distinct pf.ID , pf.status as status from /portfolio pf where pf.ID > 100 order by pf.ID, status
...
Replicated Region query:
...
For replicated executing the transformed query should yield the correct results.
...
Partitioned Region:
...
Should work without change (assuming distinct order by is supported for PR)
Implementation for queries containing only group by clause & projection containing atleast 1 at least one aggregate function & zero or more columns
select pf.status,
...
AVG(pf.ID) from /portfolio pf where pf.ID > 0 group by pf.status
Replicated Region query:
...
Get the projected rows ( by stripping the aggregate functions)
...
Sort the projected rows based on the Group By Columns
...
For each row of sorted data , feed the row to the aggregate functions in the projection & maintain the aggregate.
...
Prepare the resultset at the end of complete row scan.
Partitioned Region query:
For each Bucket:
...
Get the projected rows ( by stripping the aggregate functions)
...
Sort the projected rows based on the Group By Columns
...
For each sorted row , feed the row to the aggregate functions in the projection & maintain the aggregate.
...
Prepare the resultset at the end of complete row scan.
...
For the aggregate function AVG ,
...
it should also keep track of the number of rows used to produce the result of AVG.
...
For all buckets present locally
...
The result of each bucket needs to be merged locally, to
...
get the resultant resultset. The group by columns fetched in the projection attribute would be used to merge & calculate the resultant resultset
...
For the query node
...
Resultset from each node needs to be merged in the same manner as for all buckets merged locally.
...
Final resultset computed
Modification in current Querying Engine and things to consider:
...
The current OQL engine needs to support non distinct order by clause. The order by comparator attributes which map to run time iterators, need to map to projection attributes, so that order by clause, can be applied on the query node without firing another order by query. The The results from individuals bucket nodes need to be N - way merged, as the results from individual nodes comes sorted.
In case of PR, for certain types of aggregate functions, the the implementation of the aggregate function on bucket nodes, would defer from the implementation of the aggregate functions for the query node.
For egE.g.: select col1, avg( col2) from /portfolios group by col1.
The computed average value for each bucket , cannot be just fed in the avg() function on the query node. The avg() function on each bucket node needs to pass , the sum as well as the number of elements , to the query node. The query node avg function would take these two as input & compute a final value of avg.
For the queries which involve distinct keyword, also requires difference in implementation on the bucket node & query nodes.
For eg:
For eg: E.g.: select col1, sum( distinct col2) from /portfolios group by col1.
Here it will not be possible to pass the individual sum calculated from each of the bucket nodes, to be passed to the query node to compute the final sum. The reason being that we need to sum only distinct values of col2 and it is possible two buckets have the same value & so only one such value should be added to sum. In such case, the sum() function on function on the bucket node needs to return a Set of the values seen, rather than sum of those values.
In built aggregate functions:
MAX:
...
Usage : MAX(Expression)
example example : MAX(pf.ID) : The expression must evaluate to a java.lang.Comparable type
MIN:
...
Usage : MIN(Expression)
example example : MIN(pf.ID) : The expression must evaluate to a java.lang.Comparable type
AVG:
...
Usage : AVG(Expression)
example example : AVG AVG(pf.ID) : The expression must evaluate to java.lang.Number type. Number type.
4) AVG: Usage : AVG ( distinct expression)
5) SUM: Usage Usage : SUM(Expression)
example example : SUM SUM(pf.ID) : The expression must evaluate to java.lang.Number typeNumber type.
6) SUM: Usage : SUM( distinct expression)
7) COUNT(*): Already supported ( I think)
COUNT 8) Count ( expression | column) 9
) Count COUNT ( distinct expression | column)
User Defined Aggregate Function (UDA):
GemFire OQL Engine will provide the following interface to be implemented by the user
com.gemstone.gemfire.cache.query.Aggregator
The Aggregator interface will have following methods :
public interface Aggregator {
/**
...
* Accumulate the next scalar value
*
* @param value
...
* @param value
*/
...
public void accumulate(Object value);
...
/**
...
* Initialize the Aggregator
...
*/
...
public void init();
...
/**
*
* @return Return the result scalar value
*/
...
* @return Return the result scalar value
*/
public Object terminate();
...
/**
...
* Merges the incoming aggregator from bucket nodes with the resultant aggregator
...
* on the query node
*
* @param otherAggregator
...
* @param otherAggregator
*/
...
public void merge(Aggregator otherAggregator);
}
There should be a zero arg constructor present in the implementing class.
Example of query using built in aggregateselect pf
select pf.ID , SUM( pos.mktValue ) from /portfolios pf , pf.positions pos group by pf.ID
The above query will give the total market value of all the positions for a portfolio identified by its ID
Note:
...
As a first cut , we will make it mandatory for the group by column to be part of projection attribute ( Will see if we can make implement in a reasonable period the code to relax this constraint. i.e group by columns not be part of projection columns)
...
Proposed
...
Syntax for using UDA:
Creation of UDA
Using API:
QueryService qs = CacheUtils.getQueryService();
...
qs.createUDA(“udaAlias”, "com.gemstone.gemfire.cache.query.dunit.UDACreationDUnitTest$SumUDA");
XML based creation:
<uda-manager>
<uda <uda name="uda2" class="com.gemstone.gemfire.cache.query.dunit.UDACreationDUnitTest$UDACLass2"/>
<uda <uda name="uda3" class="com.gemstone.gemfire.cache.query.dunit.UDACreationDUnitTest$UDACLass3"/>
<</uda-manager>
cache-9.0 xsd
<xsd:element maxOccurs="1" minOccurs="0" name="uda-manager">
<xsd <xsd:complexType>
<xsd <xsd:sequence>
<xsd <xsd:element maxOccurs="unbounded" minOccurs="0" name="uda">
<xsd <xsd:complexType>
<xsd <xsd:attribute name="name" type="xsd:string" use="required" />
<xsd <xsd:attribute name="class" type="xsd:string" use="required" />
< </xsd:complexType>
< </xsd:element>
< </xsd:sequence>
< </xsd:complexType>
<</xsd:element> Example of using UDA , after definingelement>
Using UDA in Query:
String queryStr = "select p.status , myUDA(p.ID) from /portfolio p where p.ID > 0 group by p.status order by p.status";
This will require us to maintain a registry of the aggregate functions defined & will also have to ensure that it gets created on newly joining nodes.We piggy back on the infra through which the OQL indexes get created on the peers & I assume that we can also create on newly joining nodes.
UDA Schema exchange:
The UDA schema/definitions are exchanged between cluster nodes via profile exchange mechanism.