@Override public void endWindow(){
  if (columnGroupIndexes.size() == 0) {
    rows=new ArrayList<Map<String,Object>>();
    return;
  }
  HashMap<MultiKeyCompare,ArrayList<Map<String,Object>>> groups=new HashMap<MultiKeyCompare,ArrayList<Map<String,Object>>>();
  for (  Map<String,Object> row : rows) {
    MultiKeyCompare key=new MultiKeyCompare();
    for (    ColumnIndex index : columnGroupIndexes) {
      key.addCompareKey(row.get(index.getColumn()));
    }
    ArrayList<Map<String,Object>> subRows;
    if (groups.containsKey(key)) {
      subRows=groups.get(key);
    }
 else {
      subRows=new ArrayList<Map<String,Object>>();
      groups.put(key,subRows);
    }
    subRows.add(row);
  }
  for (  Map.Entry<MultiKeyCompare,ArrayList<Map<String,Object>>> entry : groups.entrySet()) {
    ArrayList<Map<String,Object>> subRows=entry.getValue();
    Map<String,Object> result=new HashMap<String,Object>();
    for (    ColumnIndex index : columnGroupIndexes) {
      index.filter(subRows.get(0),result);
    }
    for (    FunctionIndex aggregate : aggregates) {
      try {
        aggregate.filter(subRows,result);
      }
 catch (      Exception e) {
        e.printStackTrace();
      }
    }
    boolean isValidHaving=true;
    for (    HavingCondition condition : havingConditions) {
      try {
        isValidHaving&=condition.isValidAggregate(subRows);
      }
 catch (      Exception e) {
        e.printStackTrace();
        return;
      }
    }
    if (isValidHaving) {
      outport.emit(result);
    }
  }
  rows=new ArrayList<Map<String,Object>>();
}

@ Override public void endWindow ( ) { if ( columnGroupIndexes . size ( ) == 0 ) { rows = new ArrayList < Map < String , Object > > ( ) ; return ; } HashMap < MultiKeyCompare , ArrayList < Map < String , Object > > > groups = new HashMap < MultiKeyCompare , ArrayList < Map < String , Object > > > ( ) ; for ( Map < String , Object > row : rows ) { MultiKeyCompare key = new MultiKeyCompare ( ) ; for ( ColumnIndex index : columnGroupIndexes ) { key . addCompareKey ( row . get ( index . getColumn ( ) ) ) ; } ArrayList < Map < String , Object > > subRows ; if ( groups . containsKey ( key ) ) { subRows = groups . get ( key ) ; } else { subRows = new ArrayList < Map < String , Object > > ( ) ; groups . put ( key , subRows ) ; } subRows . add ( row ) ; } for ( Map . Entry < MultiKeyCompare , ArrayList < Map < String , Object > > > entry : groups . entrySet ( ) ) { ArrayList < Map < String , Object > > subRows = entry . getValue ( ) ; Map < String , Object > result = new HashMap < String , Object > ( ) ; for ( ColumnIndex index : columnGroupIndexes ) { index . filter ( subRows . get ( 0 ) , result ) ; } for ( FunctionIndex aggregate : aggregates ) { try { aggregate . filter ( subRows , result ) ; } catch ( Exception e ) { e . printStackTrace ( ) ; } } boolean isValidHaving = true ; for ( HavingCondition condition : havingConditions ) { try { isValidHaving &= condition . isValidAggregate ( subRows ) ; } catch ( Exception e ) { e . printStackTrace ( ) ; return ; } } if ( isValidHaving ) { outport . emit ( result ) ; } } rows = new ArrayList < Map < String , Object > > ( ) ; }
create aggregate at end window .
