Expose aggregate functionality in dcdbquery

tools/dcdbquery/query.cpp

@@ -103,9 +103,29 @@ void DCDBQuery::genOutput(std::list<DCDB::SensorDataStoreReading> &results, quer
     for (queryMap_t::iterator it=start; it!=stop; it++) {
 	switch(it->second.operation) {
 	    case DCDB_OP_NONE:
-	    case DCDB_OP_WINTERMUTE:
-		std::cout << ",Value";
+	    case DCDB_OP_WINTERMUTE: {
+		switch(it->second.aggregate) {
+		    case DCDB::AGGREGATE_MIN:
+			std::cout << ",min";
+			break;
+		    case DCDB::AGGREGATE_MAX:
+			std::cout << ",max";
+			break;
+		    case DCDB::AGGREGATE_AVG:
+			std::cout << ",avg";
+			break;
+		    case DCDB::AGGREGATE_SUM:
+			std::cout << ",sum";
+			break;
+		    case DCDB::AGGREGATE_COUNT:
+			std::cout << ",count";
+			break;
+		    default:
+			std::cout << ",Value";
+			break;
+		}
 		break;
+	    }
 	    case DCDB_OP_DELTA:
 		std::cout << ",Delta";
 		break;
@@ -253,7 +273,7 @@ void DCDBQuery::parseSensorSpecification(const std::string sensor, std::string&
 	modifierStr = match[2].str();
     }
     
-    queryCfg = { 1.0, DCDB::Unit_None, DCDB_OP_NONE};
+    queryCfg = { 1.0, DCDB::Unit_None, DCDB_OP_NONE, DCDB::AGGREGATE_NONE};
     if (functName.length() == 0) {
 	queryCfg.operation = DCDB_OP_NONE;
     } else if (boost::iequals(functName, "delta")) {
@@ -264,6 +284,18 @@ void DCDBQuery::parseSensorSpecification(const std::string sensor, std::string&
 	queryCfg.operation = DCDB_OP_DERIVATIVE;
     } else if (boost::iequals(functName, "integral")) {
 	queryCfg.operation = DCDB_OP_INTEGRAL;
+    } else if (boost::iequals(functName, "min")) {
+	queryCfg.aggregate = DCDB::AGGREGATE_MIN;
+    } else if (boost::iequals(functName, "max")) {
+	queryCfg.aggregate = DCDB::AGGREGATE_MAX;
+    } else if (boost::iequals(functName, "avg")) {
+	queryCfg.aggregate = DCDB::AGGREGATE_AVG;
+    } else if (boost::iequals(functName, "min")) {
+	queryCfg.aggregate = DCDB::AGGREGATE_SUM;
+    } else if (boost::iequals(functName, "sum")) {
+	queryCfg.aggregate = DCDB::AGGREGATE_MIN;
+    } else if (boost::iequals(functName, "count")) {
+	queryCfg.aggregate = DCDB::AGGREGATE_COUNT;
     } else {
 	queryCfg.operation = DCDB_OP_WINTERMUTE;
 	queryCfg.wintermuteOp = functName;
@@ -368,6 +400,7 @@ void DCDBQuery::execute() {
     std::string prevSensorName;
     for (auto q: queries) {
 	if (q.first.name != prevSensorName) {
+	    // Find all queries for the same sensor
 	    std::pair<queryMap_t::iterator, queryMap_t::iterator> range = queries.equal_range(q.first);
 	    
 	    /* Base scaling factor and unit of the public sensor */
@@ -376,10 +409,38 @@ void DCDBQuery::execute() {
 	    
 	    std::list<DCDB::SensorDataStoreReading> results;
 	    DCDB::Sensor sensor(connection, q.first);
-	    sensor.query(results, start_ts, end_ts, DCDB::AGGREGATE_NONE);
-	    
-	    genOutput(results, range.first, range.second);
 	    
+	    // Query aggregates first
+	    auto it=range.first;
+	    while(it!=range.second) {
+		if (it->second.aggregate != DCDB::AGGREGATE_NONE) {
+		    sensor.query(results, start_ts, end_ts, it->second.aggregate);
+		    if (results.size() > 0) {
+			genOutput(results, it, std::next(it));
+			results.clear();
+			if (it == range.first) {
+			    range.first = std::next(it);
+			}
+			it = queries.erase(it);
+			continue;
+		    }
+		}
+		it++;
+	    }
+
+	    // Query raw values next
+	    for (auto it=range.first; it!=range.second; it++) {
+		if (it->second.aggregate == DCDB::AGGREGATE_NONE) {
+		    sensor.query(results, start_ts, end_ts, DCDB::AGGREGATE_NONE);
+		    break;
+		}
+	    }
+	    if (results.size() > 0) {
+		genOutput(results, range.first, range.second);
+		results.clear();
+	    }
+
+
 	    prevSensorName = q.first.name;
 	}
     }

tools/dcdbquery/query.h

@@ -59,6 +59,7 @@ typedef struct queryConfig {
     double scalingFactor;
     DCDB::Unit unit;
     DCDB_OP_TYPE operation;
+    DCDB::QueryAggregate aggregate;
     std::string wintermuteOp;
 } queryConfig_t;
 typedef std::multimap<DCDB::PublicSensor, queryConfig_t> queryMap_t;