Allow group by if variable present in table
@@ -437,11 +437,13 @@ public class AggregationRuntime implements MemoryCalculable {
queryGroupByList.remove(timestampVariable);
}
for (Variable queryGroupBy : queryGroupByList) {
- if (groupByVariablesList.contains(queryGroupBy.getAttributeName())) {
String referenceId = queryGroupBy.getStreamId();
- if (aggReferenceId == null || aggReferenceId.equalsIgnoreCase(referenceId)) {
+ if (aggReferenceId == null) {
+ if (tableAttributesNameList.contains(queryGroupBy.getAttributeName())) {
groupByList.add(queryGroupBy);
}
+ } else if (aggReferenceId.equalsIgnoreCase(referenceId)) {
+ groupByList.add(queryGroupBy);
}
}
// If query group bys are based on joining stream
@@ -643,9 +645,15 @@ public class AggregationRuntime implements MemoryCalculable {
}
for (; i < baseAggregatorBeginIndex; i++) {
- OutputAttribute outputAttribute = new OutputAttribute(attributeList.get(i).getName(),
+ OutputAttribute outputAttribute;
+ Variable variable = new Variable(attributeList.get(i).getName());
+ if (queryGroupByNames.contains(variable.getAttributeName())) {
+ outputAttribute = new OutputAttribute(variable);
+ } else {
+ outputAttribute = new OutputAttribute(attributeList.get(i).getName(),
Expression.function("incrementalAggregator", "last",
new Variable(attributeList.get(i).getName()), maxVariable));
+ }
selectorList.add(outputAttribute);
}
