code: @Override
    public AggregationResult aggregate(List<AggregateField> fields) throws DDFException {
  
      String tableName = this.getDDF().getTableName();
  
      String sqlCmd = AggregateField.toSql(fields, tableName);
      mLog.info("SQL Command: " + sqlCmd);
      int numUnaggregatedFields = 0;
  
      for (AggregateField field : fields) {
        if (!field.isAggregated()) numUnaggregatedFields++;
      }
  
      try {
        List<String> result = this.getManager().sql(sqlCmd, false).getRows();
        return AggregationResult.newInstance(result, numUnaggregatedFields);
  
      } catch (Exception e) {
        e.printStackTrace();
        mLog.error(e.getMessage());
        throw new DDFException("Unable to query from " + tableName, e);
      }
    }
docs: Performs the equivalent of a SQL aggregation statement like "SELECT year, month, AVG(depdelay), MIN(arrdelay) FROM
  airline GROUP BY year, month"
  
  @param fields {@link AggregateField}s representing a list of column specs, some of which may be aggregated, while other
  non-aggregated fields are the GROUP BY keys
  @return
  @throws DDFException