code: private void mergeDataPoints() {
      // Merge sorted spans together
      final long merge_start = DateTime.nanoTime();
      for (final List<KeyValue> kvs : kv_map.values()) {
        if (kvs == null || kvs.isEmpty()) {
          LOG.warn("Found a key value list that was null or empty");
          continue;
        }
        
        for (final KeyValue kv : kvs) {
          if (kv == null) {
            LOG.warn("Found a key value item that was null");
            continue;
          }
          if (kv.key() == null) {
            LOG.warn("A key for a kv was null");
            continue;
          }
  
          Span datapoints = spans.get(kv.key());
          if (datapoints == null) {
            datapoints = RollupQuery.isValidQuery(rollup_query) ?
                new RollupSpan(tsdb, this.rollup_query) : new Span(tsdb);
            spans.put(kv.key(), datapoints);
          }
  
          if (annotation_map.containsKey(kv.key())) {
            for (final Annotation note: annotation_map.get(kv.key())) {
              datapoints.getAnnotations().add(note);
            }
            annotation_map.remove(kv.key());
          }
          try {  
            datapoints.addRow(kv);
          } catch (RuntimeException e) {
            LOG.error("Exception adding row to span", e);
            throw e;
          }
        }
      }
       
      kv_map.clear();
  
      for (final byte[] key : annotation_map.keySet()) {
        Span datapoints = spans.get(key);
        if (datapoints == null) {
          datapoints = new Span(tsdb);
          spans.put(key, datapoints);
        }
  
        for (final Annotation note: annotation_map.get(key)) {
          datapoints.getAnnotations().add(note);
        }
      }
      
      annotation_map.clear();
    }
docs: Called once all of the scanners have reported back in to record our
  latency and merge the results into the spans map. If there was an exception
  stored then we'll return that instead.