Map<String, String> existingStat, Map<String, String> incomingStat)
throws HelixException
{
if (existingStat == null)
{
throw new HelixException("existing stat for merge is null");
}
if (incomingStat == null)
{
throw new HelixException("incoming stat for merge is null");
}
// get agg type and arguments, then get agg object
String aggTypeStr = ExpressionParser.getAggregatorStr(statName);
String[] aggArgs = ExpressionParser.getAggregatorArgs(statName);
Aggregator agg = ExpressionParser.getAggregator(aggTypeStr);
// XXX: some of below lines might fail with null exceptions
// get timestamps, values out of zk maps
String existingTime = existingStat.get(TIMESTAMP_NAME);
String existingVal = existingStat.get(VALUE_NAME);
String incomingTime = incomingStat.get(TIMESTAMP_NAME);
String incomingVal = incomingStat.get(VALUE_NAME);
// parse values into tuples, if the values exist. else, tuples are null
Tuple<String> existingTimeTuple = (existingTime != null) ? Tuple
.fromString(existingTime) : null;
Tuple<String> existingValueTuple = (existingVal != null) ? Tuple
.fromString(existingVal) : null;
Tuple<String> incomingTimeTuple = (incomingTime != null) ? Tuple
.fromString(incomingTime) : null;
Tuple<String> incomingValueTuple = (incomingVal != null) ? Tuple
.fromString(incomingVal) : null;
// dp merge
agg.merge(existingValueTuple, incomingValueTuple, existingTimeTuple,
incomingTimeTuple, aggArgs);
// put merged tuples back in map
Map<String, String> mergedMap = new HashMap<String, String>();
if (existingTimeTuple.size() == 0)
{
throw new HelixException("merged time tuple has size zero");
}
if (existingValueTuple.size() == 0)
{
throw new HelixException("merged value tuple has size zero");
}
mergedMap.put(TIMESTAMP_NAME, existingTimeTuple.toString());
mergedMap.put(VALUE_NAME, existingValueTuple.toString());
return mergedMap;