closes #164 - Limit memory consumed by DownsampleIterator and return iteratively on time bucket boundaries

.gitignore

@@ -6,5 +6,4 @@ target/
 *.iml
 logs/
 server/pki/
-client/src/main/python/setup.py
 

analytics/pom.xml

@@ -23,7 +23,7 @@ under the License.
     <parent>
         <groupId>timely</groupId>
         <artifactId>timely</artifactId>
-        <version>0.0.7-SNAPSHOT</version>
+        <version>0.0.6-SNAPSHOT</version>
     </parent>
     <properties>
         <flink.version>1.1.4</flink.version>

client/pom.xml

@@ -5,11 +5,11 @@
     <parent>
         <groupId>timely</groupId>
         <artifactId>timely</artifactId>
-        <version>0.0.7-SNAPSHOT</version>
+        <version>0.0.6-SNAPSHOT</version>
     </parent>
     <properties>
         <!-- PEP 440 compliant version gets supstituted into src/main/python/template/setup.py -->
-        <python.client.version>0.0.7.dev0</python.client.version>
+        <python.client.version>0.0.6.dev0</python.client.version>
     </properties>
     <dependencies>
         <dependency>

collectd-nsq-plugin/pom.xml

@@ -4,7 +4,7 @@
     <parent>
         <groupId>timely</groupId>
         <artifactId>timely</artifactId>
-        <version>0.0.7-SNAPSHOT</version>
+        <version>0.0.6-SNAPSHOT</version>
     </parent>
     <artifactId>collectd-nsq-plugin</artifactId>
     <dependencies>

collectd-plugin-parent/pom.xml

@@ -4,7 +4,7 @@
     <parent>
         <groupId>timely</groupId>
         <artifactId>timely</artifactId>
-        <version>0.0.7-SNAPSHOT</version>
+        <version>0.0.6-SNAPSHOT</version>
     </parent>
     <artifactId>collectd-plugin-parent</artifactId>
     <dependencies>

collectd-timely-plugin/pom.xml

@@ -4,7 +4,7 @@
     <parent>
         <groupId>timely</groupId>
         <artifactId>timely</artifactId>
-        <version>0.0.7-SNAPSHOT</version>
+        <version>0.0.6-SNAPSHOT</version>
     </parent>
     <artifactId>collectd-timely-plugin</artifactId>
     <dependencies>

common/pom.xml

@@ -3,7 +3,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>0.0.7-SNAPSHOT</version>
+    <version>0.0.6-SNAPSHOT</version>
   </parent>
   <artifactId>timely-common</artifactId>
   <dependencies>

grafana/pom.xml

@@ -4,7 +4,7 @@
     <parent>
         <groupId>timely</groupId>
         <artifactId>timely</artifactId>
-        <version>0.0.7-SNAPSHOT</version>
+        <version>0.0.6-SNAPSHOT</version>
     </parent>
     <build>
         <plugins>

pom.xml

@@ -3,7 +3,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>0.0.7-SNAPSHOT</version>
+    <version>0.0.6-SNAPSHOT</version>
     <packaging>pom</packaging>
     <name>Timely</name>
     <description>A secure and scalable time series database based on Apache Accumulo</description>

server/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>timely</groupId>
         <artifactId>timely</artifactId>
-        <version>0.0.7-SNAPSHOT</version>
+        <version>0.0.6-SNAPSHOT</version>
     </parent>
     <dependencies>
         <dependency>

server/src/main/java/timely/adapter/accumulo/MetricAdapter.java

@@ -9,14 +9,12 @@ import org.apache.accumulo.core.data.Value;
 import org.apache.accumulo.core.security.ColumnVisibility;
 import org.apache.accumulo.core.util.ComparablePair;
 import org.apache.accumulo.core.util.Pair;
-import org.apache.hadoop.io.Text;
 import timely.model.Metric;
 import timely.model.Tag;
 import timely.model.parse.TagListParser;
 import timely.model.parse.TagParser;
 
 import java.nio.ByteBuffer;
-import java.nio.charset.Charset;
 import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
@@ -30,23 +28,12 @@ public class MetricAdapter {
     private static final PairLexicoder<String, Long> rowCoder = new PairLexicoder<>(new StringLexicoder(),
             new LongLexicoder());
 
-    private static final PairLexicoder<Long, String> colQualCoder = new PairLexicoder<>(new LongLexicoder(),
-            new StringLexicoder());
-
     private static final TagParser tagParser = new TagParser();
     private static final TagListParser tagListParser = new TagListParser();
 
     public static final ColumnVisibility EMPTY_VISIBILITY = new ColumnVisibility();
     public static final String VISIBILITY_TAG = "viz";
 
-    public static long roundTimestampToLastHour(long timestamp) {
-        return timestamp - (timestamp % 3600000);
-    }
-
-    public static long roundTimestampToNextHour(long timestamp) {
-        return timestamp - (timestamp % 3600000) + 3600000;
-    }
-
     public static Mutation toMutation(Metric metric) {
         final Mutation mutation = new Mutation(encodeRowKey(metric));
 
@@ -64,9 +51,8 @@ public class MetricAdapter {
                     .map(Tag::join)
                     .collect(Collectors.joining(","));
             // @formatter:on
-            byte[] cqBytes = encodeColQual(metric.getValue().getTimestamp(), cq);
-            mutation.put(new Text(cf.getBytes(Charset.forName("UTF-8"))), new Text(cqBytes), extractVisibility(tags),
-                    metric.getValue().getTimestamp(), extractValue(metric));
+
+            mutation.put(cf, cq, extractVisibility(tags), metric.getValue().getTimestamp(), extractValue(metric));
         }
         return mutation;
     }
@@ -101,11 +87,10 @@ public class MetricAdapter {
         // @formatter:off
         Metric.Builder builder = Metric.newBuilder()
                 .name(row.getFirst())
-                .value(k.getTimestamp(), ByteBuffer.wrap(v.get()).getDouble())
+                .value(row.getSecond(), ByteBuffer.wrap(v.get()).getDouble())
                 .tag(tagParser.parse(k.getColumnFamily().toString()));
         // @formatter:on
-        ComparablePair<Long, String> cq = colQualCoder.decode(k.getColumnQualifier().getBytes());
-        tagListParser.parse(cq.getSecond()).forEach(builder::tag);
+        tagListParser.parse(k.getColumnQualifier().toString()).forEach(builder::tag);
         if (includeVizTag && k.getColumnVisibility().getLength() > 0) {
             tagListParser.parse("viz=" + k.getColumnVisibility().toString()).forEach(builder::tag);
         }
@@ -121,16 +106,7 @@ public class MetricAdapter {
     }
 
     public static byte[] encodeRowKey(Metric metric) {
-        // round timestamp to hour for scan efficiency and compression
-        return encodeRowKey(metric.getName(), roundTimestampToLastHour(metric.getValue().getTimestamp()));
-    }
-
-    public static byte[] encodeColQual(Long timestamp, String colQual) {
-        return colQualCoder.encode(new ComparablePair<>(timestamp, colQual));
-    }
-
-    public static Pair<Long, String> decodeColQual(byte[] colQual) {
-        return colQualCoder.decode(colQual);
+        return encodeRowKey(metric.getName(), metric.getValue().getTimestamp());
     }
 
     public static Pair<String, Long> decodeRowKey(Key k) {

server/src/main/java/timely/api/request/MetricRequest.java

@@ -54,11 +54,7 @@ public class MetricRequest implements TcpRequest, HttpPostRequest, WebSocketRequ
     @Override
     public void parse(String line) {
         LOG.trace("Parsing Line: {}", line);
-        try {
-            metric = metricParser.parse(line);
-        } catch (Exception e) {
-            LOG.error("Error parsing metric: {}", line);
-        }
+        metric = metricParser.parse(line);
     }
 
     @Override

server/src/main/java/timely/sample/iterators/DownsampleIterator.java

@@ -14,7 +14,7 @@ import org.apache.accumulo.core.iterators.IteratorEnvironment;
 import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
 import org.apache.accumulo.core.iterators.WrappingIterator;
 
-import org.slf4j.LoggerFactory;
+import org.apache.log4j.Logger;
 import timely.adapter.accumulo.MetricAdapter;
 import timely.model.Metric;
 import timely.model.ObjectSizeOf;
@@ -26,7 +26,6 @@ import timely.sample.DownsampleFactory;
 
 public class DownsampleIterator extends WrappingIterator {
 
-    private static final org.slf4j.Logger LOG = LoggerFactory.getLogger(DownsampleIterator.class);
     private static final String START = "downsample.start";
     private static final String END = "downsample.end";
     private static final String PERIOD = "downsample.period";
@@ -39,7 +38,100 @@ public class DownsampleIterator extends WrappingIterator {
     private long end;
     private long period;
     private Key last;
-    private DownsampleMemoryEstimator memoryEstimator = null;
+    private MemoryEstimator memoryEstimator = null;
+
+    static public class MemoryEstimator {
+
+        private boolean newBucket = false;
+        private long start;
+        private long period;
+        private long startOfCurrentBucket;
+        private long bucketsCompleted = 0;
+        private long bytesPerBucket = 0;
+        boolean highVolumeBuckets = false;
+        long maxDownsampleMemory = 0; // max aggregation memory (bytes) before
+                                      // current batch is returned (after
+                                      // bucket is complete)
+        LinkedList<Long> percentageChecks = new LinkedList<>();
+
+        public void reset() {
+            newBucket = false;
+            bucketsCompleted = 0;
+            bytesPerBucket = 0;
+            percentageChecks.add(5l);
+            percentageChecks.add(25l);
+            percentageChecks.add(50l);
+            percentageChecks.add(75l);
+        }
+
+        public MemoryEstimator(long maxDownsampleMemory, long start, long period) {
+            this.maxDownsampleMemory = maxDownsampleMemory;
+            this.start = start;
+            this.period = period;
+            this.startOfCurrentBucket = this.start;
+            reset();
+        }
+
+        public double getMemoryUsedPercentage() {
+            return (bucketsCompleted * bytesPerBucket) / (double) maxDownsampleMemory * 100;
+        }
+
+        public long getBucketsCompleted() {
+            return bucketsCompleted;
+        }
+
+        public long getBytesPerBucket() {
+            return bytesPerBucket;
+        }
+
+        public boolean isHighVolumeBuckets() {
+            return highVolumeBuckets;
+        }
+
+        public boolean shouldReturnBasedOnMemoryUsage(long timestamp, Object value) {
+
+            boolean shouldReturn = false;
+            if (maxDownsampleMemory >= 0) {
+                sample(timestamp);
+                if (isNewBucket()) {
+                    bucketsCompleted++;
+                    double memoryUsedPercentage = getMemoryUsedPercentage();
+                    if (memoryUsedPercentage >= 100) {
+                        shouldReturn = true;
+                    } else {
+                        boolean checkMemoryNow = false;
+                        Long check = percentageChecks.peek();
+                        if (check != null && memoryUsedPercentage >= check) {
+                            checkMemoryNow = true;
+                            percentageChecks.removeFirst();
+                        }
+                        // recalculate bytesPerBucket
+                        if (bytesPerBucket == 0 || highVolumeBuckets || checkMemoryNow) {
+                            long memoryUsed = ObjectSizeOf.Sizer.getObjectSize(value);
+                            bytesPerBucket = memoryUsed / bucketsCompleted;
+                        }
+                        // bucket average greater than 10% of max
+                        highVolumeBuckets = (bytesPerBucket / (double) maxDownsampleMemory) >= 0.1;
+                        shouldReturn = false;
+                    }
+                }
+            }
+            return shouldReturn;
+        }
+
+        public boolean isNewBucket() {
+            return newBucket;
+        }
+
+        private void sample(long timestamp) {
+            if (timestamp >= (startOfCurrentBucket + period)) {
+                newBucket = true;
+                startOfCurrentBucket = timestamp - ((timestamp - start) % period);
+            } else {
+                newBucket = false;
+            }
+        }
+    }
 
     @SuppressWarnings("unchecked")
     @Override
@@ -54,7 +146,7 @@ public class DownsampleIterator extends WrappingIterator {
         if (options.containsKey(MAX_DOWNSAMPLE_MEMORY)) {
             maxDownsampleMemory = Long.parseLong(options.get(MAX_DOWNSAMPLE_MEMORY));
         }
-        memoryEstimator = new DownsampleMemoryEstimator(maxDownsampleMemory, start, period);
+        memoryEstimator = new MemoryEstimator(maxDownsampleMemory, start, period);
 
         String aggClassname = options.get(AGGCLASS);
         Class<?> aggClass;
@@ -70,18 +162,12 @@ public class DownsampleIterator extends WrappingIterator {
     public boolean hasTop() {
 
         if (super.hasTop()) {
-            Key topKey = super.getTopKey();
-            Value topValue = super.getTopValue();
-            Metric metric = MetricAdapter.parse(topKey, topValue);
-            LOG.trace("entering hasTop() for metric=" + metric.toString());
             while (super.hasTop()) {
-                topKey = super.getTopKey();
-                topValue = super.getTopValue();
-                metric = MetricAdapter.parse(topKey, topValue);
+                Key topKey = super.getTopKey();
+                Value topValue = super.getTopValue();
+                Metric metric = MetricAdapter.parse(topKey, topValue);
                 long timestamp = metric.getValue().getTimestamp();
                 if (memoryEstimator.shouldReturnBasedOnMemoryUsage(timestamp, value)) {
-                    LOG.trace("returning current values - memory usage > " + memoryEstimator.maxDownsampleMemory
-                            + " for metric=" + metric.toString());
                     break;
                 }
                 last = topKey;
@@ -102,7 +188,6 @@ public class DownsampleIterator extends WrappingIterator {
                     throw new RuntimeException("Downstream next() failed", e);
                 }
             }
-            LOG.trace("exiting hasTop() for metric=" + metric.toString());
             return last != null;
         } else {
             return false;

server/src/main/java/timely/sample/iterators/DownsampleMemoryEstimator.java

@@ -1,101 +0,0 @@
-package timely.sample.iterators;
-
-import timely.model.ObjectSizeOf;
-
-import java.util.LinkedList;
-
-public class DownsampleMemoryEstimator {
-
-    private boolean newBucket = false;
-    private long start;
-    private long period;
-    private long startOfCurrentBucket;
-    private long bucketsCompleted = 0;
-    private long bytesPerBucket = 0;
-    boolean highVolumeBuckets = false;
-    long maxDownsampleMemory = 0; // max aggregation memory (bytes) before
-                                  // current batch is returned (after bucket is
-                                  // complete)
-    LinkedList<Long> percentageChecks = new LinkedList<>();
-
-    public void reset() {
-        newBucket = false;
-        bucketsCompleted = 0;
-        bytesPerBucket = 0;
-        percentageChecks.clear();
-        percentageChecks.add(1l);
-        percentageChecks.add(5l);
-        percentageChecks.add(25l);
-        percentageChecks.add(50l);
-        percentageChecks.add(75l);
-        percentageChecks.add(95l);
-    }
-
-    public DownsampleMemoryEstimator(long maxDownsampleMemory, long start, long period) {
-        this.maxDownsampleMemory = maxDownsampleMemory;
-        this.start = start;
-        this.period = period;
-        this.startOfCurrentBucket = this.start;
-        reset();
-    }
-
-    public double getMemoryUsedPercentage() {
-        return (bucketsCompleted * bytesPerBucket) / (double) maxDownsampleMemory * 100;
-    }
-
-    public long getBucketsCompleted() {
-        return bucketsCompleted;
-    }
-
-    public long getBytesPerBucket() {
-        return bytesPerBucket;
-    }
-
-    public boolean isHighVolumeBuckets() {
-        return highVolumeBuckets;
-    }
-
-    public boolean shouldReturnBasedOnMemoryUsage(long timestamp, Object value) {
-
-        boolean shouldReturn = false;
-        if (maxDownsampleMemory >= 0) {
-            sample(timestamp);
-            if (isNewBucket()) {
-                bucketsCompleted++;
-                double memoryUsedPercentage = getMemoryUsedPercentage();
-                if (memoryUsedPercentage >= 100) {
-                    shouldReturn = true;
-                } else {
-                    boolean checkMemoryNow = false;
-                    Long check = percentageChecks.peek();
-                    if (check != null && memoryUsedPercentage >= check) {
-                        checkMemoryNow = true;
-                        percentageChecks.removeFirst();
-                    }
-                    // recalculate bytesPerBucket
-                    if (bytesPerBucket < 100 || bucketsCompleted == 10 || highVolumeBuckets || checkMemoryNow) {
-                        long memoryUsed = ObjectSizeOf.Sizer.getObjectSize(value);
-                        bytesPerBucket = memoryUsed / bucketsCompleted;
-                    }
-                    // bucket average greater than 10% of max
-                    highVolumeBuckets = (bytesPerBucket / (double) maxDownsampleMemory) >= 0.1;
-                    shouldReturn = false;
-                }
-            }
-        }
-        return shouldReturn;
-    }
-
-    public boolean isNewBucket() {
-        return newBucket;
-    }
-
-    private void sample(long timestamp) {
-        if (timestamp >= (startOfCurrentBucket + period)) {
-            newBucket = true;
-            startOfCurrentBucket = timestamp - ((timestamp - start) % period);
-        } else {
-            newBucket = false;
-        }
-    }
-}

server/src/main/java/timely/store/DataStore.java

@@ -2,12 +2,9 @@ package timely.store;
 
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 
 import org.apache.accumulo.core.client.Scanner;
 
-import org.apache.accumulo.core.client.TableNotFoundException;
-import org.apache.accumulo.core.data.Range;
 import timely.model.Metric;
 import timely.api.request.timeseries.QueryRequest;
 import timely.api.request.timeseries.SearchLookupRequest;
@@ -16,7 +13,6 @@ import timely.api.response.TimelyException;
 import timely.api.response.timeseries.QueryResponse;
 import timely.api.response.timeseries.SearchLookupResponse;
 import timely.api.response.timeseries.SuggestResponse;
-import timely.model.Tag;
 
 public interface DataStore {
 
@@ -30,12 +26,6 @@ public interface DataStore {
 
     void flush() throws TimelyException;
 
-    long getAgeOffForMetric(String metricName);
-
-    List<Range> getQueryRanges(String metric, long start, long end, Set<Tag> colFamValues);
-
-    Set<Tag> getColumnFamilies(String metric, Map<String, String> tags) throws TableNotFoundException;
-
     Scanner createScannerForMetric(String sessionId, String metric, Map<String, String> tags, long startTime,
             long endTime, int lag, int scannerBatchSize, int scannerReadAhead) throws TimelyException;
 

server/src/main/java/timely/store/DataStoreImpl.java

@@ -5,9 +5,25 @@ import static org.apache.accumulo.core.conf.AccumuloConfiguration.getTimeInMilli
 import io.netty.handler.codec.http.HttpResponseStatus;
 
 import java.io.IOException;
-import java.nio.charset.Charset;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Set;
+import java.util.SortedMap;
+import java.util.Timer;
+import java.util.TimerTask;
+import java.util.TreeMap;
+import java.util.TreeSet;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
@@ -29,7 +45,11 @@ import org.apache.accumulo.core.client.TableExistsException;
 import org.apache.accumulo.core.client.TableNotFoundException;
 import org.apache.accumulo.core.client.ZooKeeperInstance;
 import org.apache.accumulo.core.client.security.tokens.PasswordToken;
-import org.apache.accumulo.core.data.*;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.PartialKey;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.Value;
 import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
 import org.apache.accumulo.core.iterators.user.RegExFilter;
 import org.apache.accumulo.core.security.Authorizations;
@@ -294,6 +314,7 @@ public class DataStoreImpl implements DataStore {
             metaCache.addAll(toCache);
         }
         try {
+
             batchWriter.get().addMutation(MetricAdapter.toMutation(metric));
             internalMetrics.incrementMetricKeysInserted(metric.getTags().size());
         } catch (MutationsRejectedException e) {
@@ -445,20 +466,13 @@ public class DataStoreImpl implements DataStore {
     @Override
     public List<QueryResponse> query(QueryRequest msg) throws TimelyException {
         List<QueryResponse> result = new ArrayList<>();
-        LOG.debug("Query request {}", msg);
         long startTs = msg.getStart();
         long endTs = msg.getEnd();
-        StringBuilder metricList = new StringBuilder();
         try {
-            long numResults = 0;
             long now = System.currentTimeMillis();
             for (SubQuery query : msg.getQueries()) {
                 Map<Set<Tag>, List<Aggregation>> allSeries = new HashMap<>();
                 String metric = query.getMetric();
-                if (metricList.length() > 0) {
-                    metricList.append(",");
-                }
-                metricList.append(metric);
                 BatchScanner scanner = connector.createBatchScanner(metricsTable, getSessionAuthorizations(msg),
                         scannerThreads);
                 try {
@@ -472,12 +486,10 @@ public class DataStoreImpl implements DataStore {
                     long endOfLastPeriod = (endDistanceFromDownSample > 0 ? endTs + downsample
                             - endDistanceFromDownSample : endTs);
 
-                    List<String> tagOrder = prioritizeTags(query.getMetric(), query.getTags());
+                    setQueryRange(scanner, metric, startOfFirstPeriod, endOfLastPeriod);
+                    List<String> tagOrder = prioritizeTags(query);
                     Map<String, String> orderedTags = orderTags(tagOrder, query.getTags());
-                    Set<Tag> colFamValues = getColumnFamilies(metric, orderedTags);
-                    List<Range> ranges = getQueryRanges(metric, startOfFirstPeriod, endOfLastPeriod, colFamValues);
-                    scanner.setRanges(ranges);
-                    setQueryColumns(scanner, metric, orderedTags, colFamValues);
+                    setQueryColumns(scanner, metric, orderedTags);
 
                     if (query.isRate()) {
                         LOG.trace("Adding rate iterator");
@@ -517,7 +529,6 @@ public class DataStoreImpl implements DataStore {
                         // Aggregation
                         Map<Set<Tag>, Aggregation> samples = AggregationIterator.decodeValue(encoded.getValue());
                         for (Entry<Set<Tag>, Aggregation> entry : samples.entrySet()) {
-                            numResults++;
                             Set<Tag> key = new HashSet<>();
                             for (Tag tag : entry.getKey()) {
                                 if (query.getTags().keySet().contains(tag.getKey())) {
@@ -537,12 +548,10 @@ public class DataStoreImpl implements DataStore {
                 // TODO groupby here?
                 long tsDivisor = msg.isMsResolution() ? 1 : 1000;
                 for (Entry<Set<Tag>, List<Aggregation>> entry : allSeries.entrySet()) {
-                    numResults += entry.getValue().size();
                     result.add(convertToQueryResponse(query, entry.getKey(), entry.getValue(), tsDivisor));
                 }
             }
-            LOG.debug("Query time: {} duration: {} metrics: {} results: {}", (System.currentTimeMillis() - now),
-                    (endTs - startTs), metricList.toString(), numResults);
+            LOG.debug("Query time: {}", (System.currentTimeMillis() - now));
             internalMetrics.addQueryResponse(result.size(), (System.currentTimeMillis() - now));
             return result;
         } catch (ClassNotFoundException | IOException | TableNotFoundException ex) {
@@ -570,9 +579,10 @@ public class DataStoreImpl implements DataStore {
      * @param query
      * @return ordered list of most specific to least specific tags in the query
      */
-    private List<String> prioritizeTags(String metric, Map<String, String> tags) {
+    private List<String> prioritizeTags(SubQuery query) {
         // trivial cases
-        if (tags == null || tags.isEmpty()) {
+        Map<String, String> tags = query.getTags();
+        if (tags.isEmpty()) {
             return Collections.emptyList();
         }
         if (tags.size() == 1) {
@@ -580,6 +590,7 @@ public class DataStoreImpl implements DataStore {
         }
         // favor tags with fewer values
         Map<String, Integer> priority = new HashMap<>();
+        String metric = query.getMetric();
         // Count matching tags
         updateMetricCounts();
         for (Entry<String, String> entry : tags.entrySet()) {
@@ -635,9 +646,8 @@ public class DataStoreImpl implements DataStore {
         return !REGEX_TEST.matcher(value).matches();
     }
 
-    public Set<Tag> getColumnFamilies(String metric, Map<String, String> requestedTags) throws TableNotFoundException {
-
-        Map<String, String> tags = (requestedTags == null) ? new LinkedHashMap<>() : requestedTags;
+    private void setQueryColumns(ScannerBase scanner, String metric, Map<String, String> tags)
+            throws TableNotFoundException, TimelyException {
         LOG.trace("Looking for requested tags: {}", tags);
         Scanner meta = connector.createScanner(metaTable, Authorizations.EMPTY);
         Text start = new Text(Meta.VALUE_PREFIX + metric);
@@ -700,35 +710,24 @@ public class DataStoreImpl implements DataStore {
         // matches
         // the pattern of the first tag in the query. The resulting set of tags
         // will be used to fetch specific columns from the metric table.
-        return expandTagValues(first, knownKeyValues);
-    }
-
-    private void setQueryColumns(ScannerBase scanner, String metric, Map<String, String> tags, Set<Tag> colFamValues)
-            throws TimelyException {
-
-        if (colFamValues.size() == 0) {
+        Set<Tag> concrete = expandTagValues(first, knownKeyValues);
+        if (concrete.size() == 0) {
             throw new TimelyException(HttpResponseStatus.BAD_REQUEST.code(), "No matching tags", "No tags were found "
                     + " that matched the submitted tags. Please fix and retry");
         }
-        LOG.trace("Found matching tags: {}", colFamValues);
-        for (Tag tag : colFamValues) {
+        LOG.trace("Found matching tags: {}", concrete);
+        for (Tag tag : concrete) {
             Text colf = new Text(tag.getKey() + "=" + tag.getValue());
             scanner.fetchColumnFamily(colf);
             LOG.trace("Fetching metric table column family: {}", colf);
         }
         // Add the regular expression to filter the other tags
         int priority = 100;
-        Iterator<Entry<String, String>> tagIter = tags.entrySet().iterator();
-        // skip over first tag which was already expanded into colFamValues
-        if (tagIter.hasNext()) {
-            tagIter.next();
-        }
-        Entry<String, String> tag = null;
         while (tagIter.hasNext()) {
-            tag = tagIter.next();
+            Entry<String, String> tag = tagIter.next();
             LOG.trace("Adding regex filter for tag {}", tag);
             StringBuffer pattern = new StringBuffer();
-            pattern.append("(^.*\\x00|.*,)");
+            pattern.append("(^|.*,)");
             pattern.append(tag.getKey());
             pattern.append("=");
             pattern.append(tag.getValue());
@@ -771,55 +770,14 @@ public class DataStoreImpl implements DataStore {
         return result;
     }
 
-    public List<Range> getQueryRanges(String metric, long start, long end, Set<Tag> colFamValues) {
-        List<Range> ranges = new ArrayList<>();
-        long beginRangeRounded = MetricAdapter.roundTimestampToLastHour(start);
-        if (colFamValues.isEmpty()) {
-            final byte[] start_row = MetricAdapter.encodeRowKey(metric, beginRangeRounded);
-            Key startKey = new Key(new Text(start_row));
-            LOG.trace("Start key for metric {} and time {} is {}", metric, beginRangeRounded, startKey.toStringNoTime());
-            final byte[] end_row = MetricAdapter.encodeRowKey(metric, beginRangeRounded);
-            Key endKey = new Key(new Text(end_row));
-            LOG.trace("End key for metric {} and time {} is {}", metric, MetricAdapter.roundTimestampToNextHour(end),
-                    endKey.toStringNoTime());
-            Range range = new Range(startKey, true, endKey, false);
-            LOG.trace("Set query range to {}", range);
-            ranges.add(range);
-        } else {
-            try {
-                long lastBeginRangeRounded = MetricAdapter.roundTimestampToLastHour(end);
-                while (beginRangeRounded <= lastBeginRangeRounded) {
-                    // use end timestamp of begin + 1 hour and one msec
-                    // except the last range where we use end + 1 msec
-                    long endRangeTimestamp = (beginRangeRounded == lastBeginRangeRounded) ? end + 1 : beginRangeRounded
-                            + (1000 * 60 * 60) + 1;
-                    long beginRangeTimestamp = (beginRangeRounded == MetricAdapter.roundTimestampToLastHour(start)) ? start
-                            : beginRangeRounded;
-                    for (Tag t : colFamValues) {
-                        final byte[] start_row = MetricAdapter.encodeRowKey(metric, beginRangeRounded);
-                        Key startKey = new Key(new Text(start_row), new Text(t.join()
-                                .getBytes(Charset.forName("UTF-8"))), new Text(MetricAdapter.encodeColQual(
-                                beginRangeTimestamp, "")), new Text(new byte[0]), beginRangeTimestamp);
-                        LOG.trace("Start key for metric {} and time {} is {}", metric, beginRangeTimestamp,
-                                startKey.toStringNoTime());
-                        final byte[] end_row = MetricAdapter.encodeRowKey(metric, beginRangeRounded);
-                        Key endKey = new Key(new Text(end_row), new Text(t.join().getBytes(Charset.forName("UTF-8"))),
-                                new Text(MetricAdapter.encodeColQual(endRangeTimestamp, "")), new Text(new byte[0]),
-                                endRangeTimestamp);
-                        LOG.trace("End key for metric {} and time {} is {}", metric, beginRangeTimestamp,
-                                endKey.toStringNoTime());
-                        Range range = new Range(startKey, true, endKey, false);
-                        LOG.trace("Set query range to {}", range);
-                        ranges.add(range);
-                    }
-                    beginRangeRounded += (1000 * 60 * 60); // add an hour
-                }
-            } catch (Exception e) {
-                LOG.error("Error creating query ranges", e);
-            }
-        }
-
-        return ranges;
+    private void setQueryRange(BatchScanner scanner, String metric, long start, long end) {
+        final byte[] start_row = MetricAdapter.encodeRowKey(metric, start);
+        LOG.trace("Start key for metric {} and time {} is {}", metric, start, start_row);
+        final byte[] end_row = MetricAdapter.encodeRowKey(metric, end);
+        LOG.trace("End key for metric {} and time {} is {}", metric, end, end_row);
+        Range range = new Range(new Text(start_row), new Text(end_row));
+        LOG.trace("Set query range to {}", range);
+        scanner.setRanges(Collections.singletonList(range));
     }
 
     private Class<? extends Aggregator> getAggregator(SubQuery query) {
@@ -880,7 +838,7 @@ public class DataStoreImpl implements DataStore {
         }
     }
 
-    public long getAgeOffForMetric(String metricName) {
+    private long getAgeOffForMetric(String metricName) {
         String age = this.ageOff.get(MetricAgeOffIterator.AGE_OFF_PREFIX + metricName);
         if (null == age) {
             return this.defaultAgeOff;
@@ -910,13 +868,23 @@ public class DataStoreImpl implements DataStore {
             if (null == metric) {
                 throw new IllegalArgumentException("metric name must be specified");
             }
-            if (tags == null) {
-                tags = new LinkedHashMap<>();
+            if (0 == startTime) {
+                startTime = (System.currentTimeMillis() - getAgeOffForMetric(metric) - 1000);
+                LOG.debug("Overriding zero start time to {} due to age off configuration", startTime);
             }
-            List<String> tagOrder = prioritizeTags(metric, tags);
-            Map<String, String> orderedTags = orderTags(tagOrder, tags);
-            Set<Tag> colFamValues = getColumnFamilies(metric, orderedTags);
-            setQueryColumns(s, metric, orderedTags, colFamValues);
+            byte[] start = MetricAdapter.encodeRowKey(metric, startTime);
+            long endTimeStamp = (endTime == 0) ? (System.currentTimeMillis() - (lag * 1000)) : endTime;
+            byte[] end = MetricAdapter.encodeRowKey(metric, endTimeStamp);
+            s.setRange(new Range(new Text(start), true, new Text(end), false));
+            SubQuery query = new SubQuery();
+            query.setMetric(metric);
+            if (null == tags) {
+                tags = Collections.emptyMap();
+            }
+            query.setTags(tags);
+            List<String> tagOrder = prioritizeTags(query);
+            Map<String, String> orderedTags = orderTags(tagOrder, query.getTags());
+            setQueryColumns(s, metric, orderedTags);
             s.setBatchSize(scannerBatchSize);
             s.setReadaheadThreshold(scannerReadAhead);
             return s;

server/src/main/java/timely/store/iterators/TimeSeriesGroupingIterator.java

@@ -231,7 +231,7 @@ public class TimeSeriesGroupingIterator extends WrappingIterator {
         while (super.hasTop()) {
             Key k = super.getTopKey();
 
-            Long newTime = k.getTimestamp();
+            Long newTime = MetricAdapter.decodeRowKey(k).getSecond();
             if (time == null) {
                 time = newTime;
             }

server/src/main/java/timely/subscription/MetricScanner.java

@@ -2,20 +2,16 @@ package timely.subscription;
 
 import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
 import io.netty.channel.ChannelHandlerContext;
-import io.netty.handler.codec.http.HttpResponseStatus;
 import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
 import io.netty.util.concurrent.ScheduledFuture;
 
 import java.lang.Thread.UncaughtExceptionHandler;
 import java.util.Iterator;
-import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
-import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.accumulo.core.client.Scanner;
-import org.apache.accumulo.core.client.TableNotFoundException;
 import org.apache.accumulo.core.data.Key;
 import org.apache.accumulo.core.data.Range;
 import org.apache.accumulo.core.data.Value;
@@ -29,7 +25,6 @@ import timely.api.response.MetricResponse;
 import timely.api.response.MetricResponses;
 import timely.api.response.TimelyException;
 import timely.model.Metric;
-import timely.model.Tag;
 import timely.store.DataStore;
 import timely.util.JsonUtil;
 
@@ -50,16 +45,11 @@ public class MetricScanner extends Thread implements UncaughtExceptionHandler {
     private final int lag;
     private final String subscriptionId;
     private final String metric;
-    private long startTime;
-    private long endTime;
+    private final long endTime;
     private final Subscription subscription;
     private MetricResponses responses = new MetricResponses();
     private final ScheduledFuture<?> flusher;
     private final int subscriptionBatchSize;
-    private List<Range> ranges = null;
-    private Iterator<Range> rangeItr = null;
-    private DataStore store = null;
-    private Set<Tag> colFamValues = null;
 
     public MetricScanner(Subscription sub, String subscriptionId, String sessionId, DataStore store, String metric,
             Map<String, String> tags, long startTime, long endTime, long delay, int lag, ChannelHandlerContext ctx,
@@ -67,34 +57,13 @@ public class MetricScanner extends Thread implements UncaughtExceptionHandler {
             throws TimelyException {
         this.setDaemon(true);
         this.setUncaughtExceptionHandler(this);
-        this.store = store;
         this.subscription = sub;
         this.ctx = ctx;
         this.lag = lag;
         this.metric = metric;
-        this.startTime = startTime;
         this.endTime = endTime;
-        this.scanner = this.store.createScannerForMetric(sessionId, metric, tags, startTime, endTime, lag,
-                scannerBatchSize, scannerReadAhead);
-
-        if (0 == startTime) {
-            this.startTime = (System.currentTimeMillis() - this.store.getAgeOffForMetric(metric) - 1000);
-            LOG.debug("Overriding zero start time to {} due to age off configuration", startTime);
-        }
-        long endTimeStamp = (endTime == 0) ? (System.currentTimeMillis() - (lag * 1000)) : endTime;
-
-        try {
-            this.colFamValues = this.store.getColumnFamilies(metric, tags);
-            ranges = this.store.getQueryRanges(metric, this.startTime, endTimeStamp, colFamValues);
-            rangeItr = ranges.iterator();
-            if (rangeItr.hasNext()) {
-                Range r = rangeItr.next();
-                this.scanner.setRange(r);
-            }
-        } catch (TableNotFoundException e) {
-            throw new TimelyException(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(), "Error in MetricScanner",
-                    e.getMessage(), e);
-        }
+        this.scanner = store.createScannerForMetric(sessionId, metric, tags, startTime, endTime, lag, scannerBatchSize,
+                scannerReadAhead);
         this.iter = scanner.iterator();
         this.delay = delay;
         this.subscriptionId = subscriptionId;
@@ -117,7 +86,6 @@ public class MetricScanner extends Thread implements UncaughtExceptionHandler {
     }
 
     public synchronized void flush() {
-        LOG.info("Flush called");
         synchronized (responses) {
             if (responses.size() > 0) {
                 try {
@@ -135,8 +103,7 @@ public class MetricScanner extends Thread implements UncaughtExceptionHandler {
     public void run() {
         Metric m = null;
         try {
-            boolean done = false;
-            while (!done && !closed) {
+            while (!closed) {
 
                 if (this.iter.hasNext()) {
                     Entry<Key, Value> e = this.iter.next();
@@ -145,55 +112,39 @@ public class MetricScanner extends Thread implements UncaughtExceptionHandler {
                         flush();
                     }
                     this.responses.addResponse(MetricResponse.fromMetric(m, this.subscriptionId));
-                } else if (rangeItr.hasNext()) {
-                    // set next range on the scanner
-                    Range r = rangeItr.next();
-                    this.scanner.close();
-                    this.scanner.setRange(r);
-                    this.iter = scanner.iterator();
                 } else if (this.endTime == 0) {
                     flush();
                     long endTimeStamp = (System.currentTimeMillis() - (lag * 1000));
+                    byte[] end = MetricAdapter.encodeRowKey(this.metric, endTimeStamp);
+                    Text endRow = new Text(end);
                     this.scanner.close();
+                    Range prevRange = this.scanner.getRange();
                     if (null == m) {
                         LOG.debug("No results found, waiting {}ms to retry with new end time {}.", delay, endTimeStamp);
                         sleepUninterruptibly(delay, TimeUnit.MILLISECONDS);
-                        ranges = this.store.getQueryRanges(metric, startTime, endTimeStamp, colFamValues);
-                        rangeItr = ranges.iterator();
-                        if (rangeItr.hasNext()) {
-                            Range r = rangeItr.next();
-                            this.scanner.setRange(r);
-                            this.iter = scanner.iterator();
-                        }
+                        this.scanner.setRange(new Range(prevRange.getStartKey().getRow(), prevRange
+                                .isStartKeyInclusive(), endRow, false));
+                        this.iter = this.scanner.iterator();
                     } else {
                         // Reset the starting range to the last key returned
                         LOG.debug("Exhausted scanner, last metric returned was {}", m);
                         LOG.debug("Waiting {}ms to retry with new end time {}.", delay, endTimeStamp);
                         sleepUninterruptibly(delay, TimeUnit.MILLISECONDS);
-                        ranges = this.store.getQueryRanges(metric, m.getValue().getTimestamp() + 1, endTimeStamp,
-                                colFamValues);
-                        rangeItr = ranges.iterator();
-                        if (rangeItr.hasNext()) {
-                            Range r = rangeItr.next();
-                            this.scanner.setRange(r);
-                            this.iter = this.scanner.iterator();
-                        } else {
-                            done = true;
-                        }
+                        this.scanner.setRange(new Range(new Text(MetricAdapter.encodeRowKey(m)), false, endRow,
+                                prevRange.isEndKeyInclusive()));
+                        this.iter = this.scanner.iterator();
                     }
                 } else {
-                    done = true;
-                }
-                if (done) {
                     LOG.debug("Exhausted scanner, sending completed message for subscription {}", this.subscriptionId);
                     final MetricResponse last = new MetricResponse();
                     last.setSubscriptionId(this.subscriptionId);
                     last.setComplete(true);
                     this.responses.addResponse(last);
                     flush();
+                    break;
                 }
             }
-        } catch (Throwable e) {
+        } catch (Exception e) {
             LOG.error("Error in metric scanner, closing.", e);
         } finally {
             close();

server/src/main/java/timely/util/TimelyMetricsFormatter.java

@@ -15,8 +15,6 @@ public class TimelyMetricsFormatter extends DefaultFormatter {
 
     private static final PairLexicoder<String, Long> rowCoder = new PairLexicoder<>(new StringLexicoder(),
             new LongLexicoder());
-    private static final PairLexicoder<Long, String> colQualCoder = new PairLexicoder<>(new LongLexicoder(),
-            new StringLexicoder());
 
     @Override
     public String next() {
@@ -27,8 +25,7 @@ public class TimelyMetricsFormatter extends DefaultFormatter {
         b.append(" ");
         b.append(e.getKey().getColumnFamily().toString());
         b.append(":");
-        ComparablePair<Long, String> cq = colQualCoder.decode(e.getKey().getColumnQualifier().copyBytes());
-        b.append(cq.getFirst()).append("\\x00").append(cq.getSecond());
+        b.append(e.getKey().getColumnQualifier().toString());
         b.append(" ");
         b.append(e.getKey().getColumnVisibility().toString());
         if (super.isDoTimestamps()) {

server/src/test/java/timely/api/model/MetricAdapterTest.java

@@ -42,33 +42,28 @@ public class MetricAdapterTest {
         Mutation mut = MetricAdapter.toMutation(m);
 
         PairLexicoder<String, Long> rowCoder = new PairLexicoder<>(new StringLexicoder(), new LongLexicoder());
-        byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", MetricAdapter.roundTimestampToLastHour(ts)));
+        byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", ts));
         byte[] value = new byte[Double.BYTES];
         ByteBuffer.wrap(value).putDouble(2.0D);
         Assert.assertEquals(rowCoder.decode(row), rowCoder.decode(mut.getRow()));
         Assert.assertEquals(3, mut.getUpdates().size());
         ColumnUpdate up = mut.getUpdates().get(0);
-        Assert.assertEquals("tag1=value1", new String(up.getColumnFamily()));
-        PairLexicoder<Long, String> colQualCoder = new PairLexicoder<>(new LongLexicoder(), new StringLexicoder());
-        Assert.assertEquals(new String(colQualCoder.encode(new ComparablePair<>(ts, "tag2=value2,tag3=value3"))),
-                new String(up.getColumnQualifier()));
-
+        Assert.assertTrue(new String(up.getColumnFamily()).equals("tag1=value1"));
+        Assert.assertTrue(new String(up.getColumnQualifier()).equals("tag2=value2,tag3=value3"));
         Assert.assertEquals(ts, up.getTimestamp());
-        Assert.assertEquals("", new String(new String(up.getColumnVisibility())));
+        Assert.assertTrue(new String(up.getColumnVisibility()).equals(""));
         Assert.assertArrayEquals(value, up.getValue());
         ColumnUpdate up2 = mut.getUpdates().get(1);
-        Assert.assertEquals("tag2=value2", new String(up2.getColumnFamily()));
-        Assert.assertEquals(new String(colQualCoder.encode(new ComparablePair<>(ts, "tag1=value1,tag3=value3"))),
-                new String(up2.getColumnQualifier()));
+        Assert.assertTrue(new String(up2.getColumnFamily()).equals("tag2=value2"));
+        Assert.assertTrue(new String(up2.getColumnQualifier()).equals("tag1=value1,tag3=value3"));
         Assert.assertEquals(ts, up2.getTimestamp());
-        Assert.assertEquals("", new String(up2.getColumnVisibility()));
+        Assert.assertTrue(new String(up2.getColumnVisibility()).equals(""));
         Assert.assertArrayEquals(value, up.getValue());
         ColumnUpdate up3 = mut.getUpdates().get(2);
-        Assert.assertEquals("tag3=value3", new String(up3.getColumnFamily()));
-        Assert.assertEquals(new String(colQualCoder.encode(new ComparablePair<>(ts, "tag1=value1,tag2=value2"))),
-                new String(up3.getColumnQualifier()));
+        Assert.assertTrue(new String(up3.getColumnFamily()).equals("tag3=value3"));
+        Assert.assertTrue(new String(up3.getColumnQualifier()).equals("tag1=value1,tag2=value2"));
         Assert.assertEquals(ts, up3.getTimestamp());
-        Assert.assertEquals("", new String(up3.getColumnVisibility()));
+        Assert.assertTrue(new String(up3.getColumnVisibility()).equals(""));
         Assert.assertArrayEquals(value, up.getValue());
     }
 
@@ -83,16 +78,14 @@ public class MetricAdapterTest {
         Mutation mut = MetricAdapter.toMutation(m);
 
         PairLexicoder<String, Long> rowCoder = new PairLexicoder<>(new StringLexicoder(), new LongLexicoder());
-        byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", MetricAdapter.roundTimestampToLastHour(ts)));
+        byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", ts));
         byte[] value = new byte[Double.BYTES];
         ByteBuffer.wrap(value).putDouble(2.0D);
         Assert.assertEquals(rowCoder.decode(row), rowCoder.decode(mut.getRow()));
         Assert.assertEquals(1, mut.getUpdates().size());
         ColumnUpdate up = mut.getUpdates().get(0);
         Assert.assertEquals("tag1=value1", new String(up.getColumnFamily()));
-        PairLexicoder<Long, String> colQualCoder = new PairLexicoder<>(new LongLexicoder(), new StringLexicoder());
-        Assert.assertEquals(new String(colQualCoder.encode(new ComparablePair<>(ts, ""))),
-                new String(up.getColumnQualifier()));
+        Assert.assertEquals("", new String(up.getColumnQualifier()));
         Assert.assertEquals(ts, up.getTimestamp());
         Assert.assertEquals("(a&b)|(c&d)", new String(up.getColumnVisibility()));
         Assert.assertArrayEquals(value, up.getValue());
@@ -117,9 +110,8 @@ public class MetricAdapterTest {
         byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", 1000L));
         byte[] value = new byte[Double.BYTES];
         ByteBuffer.wrap(value).putDouble(2.0D);
-        PairLexicoder<Long, String> colQualCoder = new PairLexicoder<>(new LongLexicoder(), new StringLexicoder());
-        Key k = new Key(row, "tag1=value1".getBytes(), colQualCoder.encode(new ComparablePair<>(new Long(1000),
-                "tag2=value2,tag3=value3")), "(a&b)|(c&d)".getBytes(), 1000);
+        Key k = new Key(row, "tag1=value1".getBytes(), "tag2=value2,tag3=value3".getBytes(), "(a&b)|(c&d)".getBytes(),
+                1000);
         Value v = new Value(value);
         Metric m = MetricAdapter.parse(k, v);
         Assert.assertEquals("sys.cpu.user", m.getName());
@@ -138,9 +130,8 @@ public class MetricAdapterTest {
         byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", 1000L));
         byte[] value = new byte[Double.BYTES];
         ByteBuffer.wrap(value).putDouble(2.0D);
-        PairLexicoder<Long, String> colQualCoder = new PairLexicoder<>(new LongLexicoder(), new StringLexicoder());
-        Key k = new Key(row, "tag1=value1".getBytes(), colQualCoder.encode(new ComparablePair<>(new Long(1000),
-                "tag2=value2,tag3=value3")), "(a&b)|(c&d)".getBytes(), 1000);
+        Key k = new Key(row, "tag1=value1".getBytes(), "tag2=value2,tag3=value3".getBytes(), "(a&b)|(c&d)".getBytes(),
+                1000);
         Value v = new Value(value);
         Metric m = MetricAdapter.parse(k, v, true);
         Assert.assertEquals("sys.cpu.user", m.getName());

server/src/test/java/timely/sample/iterators/DownsampleIteratorTest.java

@@ -54,7 +54,6 @@ public class DownsampleIteratorTest {
             for (ColumnUpdate cu : mutation.getUpdates()) {
                 Key key = new Key(mutation.getRow(), cu.getColumnFamily(), cu.getColumnQualifier(),
                         cu.getColumnVisibility(), cu.getTimestamp());
-                System.out.println(key.toString());
                 testData1.put(key, new Value(cu.getValue()));
             }
         }
@@ -70,6 +69,7 @@ public class DownsampleIteratorTest {
     }
 
     private void createTestData2() {
+        List<List<Tag>> listOfTagVariations = new ArrayList<>();
         List<Tag> tags = Collections.singletonList(new Tag("host", "host1"));
         List<Tag> tags2 = Collections.singletonList(new Tag("host", "host2"));
 
@@ -168,7 +168,8 @@ public class DownsampleIteratorTest {
         long period = 500l;
         long sizeOfObjects = 20;
         SampleObject o = new SampleObject();
-        DownsampleMemoryEstimator memoryEstimator = new DownsampleMemoryEstimator(maxMemory, start, period);
+        DownsampleIterator.MemoryEstimator memoryEstimator = new DownsampleIterator.MemoryEstimator(maxMemory, start,
+                period);
         boolean shouldReturn = false;
         for (long x = 100; x <= 5000; x += 100) {
             long timestamp = start + x;
@@ -197,7 +198,8 @@ public class DownsampleIteratorTest {
         long period = 500l;
         long sizeOfObjects = 200;
         SampleObject o = new SampleObject();
-        DownsampleMemoryEstimator memoryEstimator = new DownsampleMemoryEstimator(maxMemory, start, period);
+        DownsampleIterator.MemoryEstimator memoryEstimator = new DownsampleIterator.MemoryEstimator(maxMemory, start,
+                period);
         boolean shouldReturn = false;
         for (long x = 100; x <= 5000; x += 100) {
             long timestamp = start + x;
@@ -279,7 +281,6 @@ public class DownsampleIteratorTest {
                 }
             }
             key = iter.getTopKey();
-            System.out.println(key.toString());
         } while (iter.hasTop());
 
         assertEquals(testData.lastKey(), key);

server/src/test/java/timely/store/iterators/RateIteratorTest.java

@@ -22,7 +22,6 @@ import timely.adapter.accumulo.MetricAdapter;
 import timely.api.request.timeseries.QueryRequest;
 import timely.model.Metric;
 import timely.model.Tag;
-import timely.store.DataStoreImpl;
 
 public class RateIteratorTest extends IteratorTestBase {
 
@@ -39,8 +38,7 @@ public class RateIteratorTest extends IteratorTestBase {
             ts += 1000;
             Metric m = new Metric("sys.cpu.user", ts, i * 1.0D, tags);
             byte[] row = MetricAdapter.encodeRowKey(m);
-            Key k = new Key(row, tags.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(ts,
-                    ""), new byte[0], ts);
+            Key k = new Key(row, tags.get(0).join().getBytes(StandardCharsets.UTF_8), new byte[0], new byte[0], ts);
             Value v = new Value(MetricAdapter.encodeValue(m.getValue().getMeasure()));
             table.put(k, v);
         }
@@ -70,8 +68,7 @@ public class RateIteratorTest extends IteratorTestBase {
             ts += 1000 + r.nextInt(100);
             Metric m = new Metric("sys.cpu.user", ts, i * 1.0D, tags);
             byte[] row = MetricAdapter.encodeRowKey(m);
-            Key k = new Key(row, tags.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(ts,
-                    ""), new byte[0], ts);
+            Key k = new Key(row, tags.get(0).join().getBytes(StandardCharsets.UTF_8), new byte[0], new byte[0], ts);
             Value v = new Value(MetricAdapter.encodeValue(m.getValue().getMeasure()));
             table.put(k, v);
         }
@@ -116,8 +113,7 @@ public class RateIteratorTest extends IteratorTestBase {
                 ts += 1000;
                 Metric m = new Metric("sys.cpu.user", ts, i * 1.0D, tags);
                 byte[] row = MetricAdapter.encodeRowKey(m);
-                Key k = new Key(row, tags.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(
-                        ts, ""), new byte[0], ts);
+                Key k = new Key(row, tags.get(0).join().getBytes(StandardCharsets.UTF_8), new byte[0], new byte[0], ts);
                 Value v = new Value(MetricAdapter.encodeValue(m.getValue().getMeasure()));
                 table.put(k, v);
             }
@@ -151,8 +147,7 @@ public class RateIteratorTest extends IteratorTestBase {
                 ts += 1000;
                 Metric m = new Metric("sys.cpu.user", ts, i * 1.0D, tags);
                 byte[] row = MetricAdapter.encodeRowKey(m);
-                Key k = new Key(row, tags.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(
-                        ts, ""), new byte[0], ts);
+                Key k = new Key(row, tags.get(0).join().getBytes(StandardCharsets.UTF_8), new byte[0], new byte[0], ts);
                 Value v = new Value(MetricAdapter.encodeValue(m.getValue().getMeasure()));
                 table.put(k, v);
             }
@@ -186,8 +181,7 @@ public class RateIteratorTest extends IteratorTestBase {
                 ts += 1000;
                 Metric m = new Metric("sys.cpu.user", ts, i * 1.0D, tags);
                 byte[] row = MetricAdapter.encodeRowKey(m);
-                Key k = new Key(row, tags.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(
-                        ts, ""), new byte[0], ts);
+                Key k = new Key(row, tags.get(0).join().getBytes(StandardCharsets.UTF_8), new byte[0], new byte[0], ts);
                 Value v = new Value(MetricAdapter.encodeValue(m.getValue().getMeasure()));
                 table.put(k, v);
             }

server/src/test/java/timely/store/iterators/TimeSeriesGroupingIteratorTest.java

@@ -6,7 +6,10 @@ import static org.junit.Assert.assertTrue;
 
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.TreeMap;
 
 import org.apache.accumulo.core.client.IteratorSetting;
 import org.apache.accumulo.core.data.Key;
@@ -40,8 +43,7 @@ public class TimeSeriesGroupingIteratorTest extends IteratorTestBase {
             ts += 1000;
             Metric m = new Metric("sys.cpu.user", ts, i * 1.0D, tags);
             byte[] row = MetricAdapter.encodeRowKey(m);
-            Key k = new Key(row, tags.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(ts,
-                    ""), new byte[0], ts);
+            Key k = new Key(row, tags.get(0).join().getBytes(StandardCharsets.UTF_8), new byte[0], new byte[0], ts);
             Value v = new Value(MetricAdapter.encodeValue(m.getValue().getMeasure()));
             table.put(k, v);
         }
@@ -90,14 +92,12 @@ public class TimeSeriesGroupingIteratorTest extends IteratorTestBase {
             ts += 1000;
             Metric m = new Metric("sys.cpu.user", ts, i * 1.0D, tags1);
             byte[] row = MetricAdapter.encodeRowKey(m);
-            Key k = new Key(row, tags1.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(ts,
-                    ""), new byte[0], ts);
+            Key k = new Key(row, tags1.get(0).join().getBytes(StandardCharsets.UTF_8), new byte[0], new byte[0], ts);
             Value v = new Value(MetricAdapter.encodeValue(m.getValue().getMeasure()));
             table.put(k, v);
             Metric m2 = new Metric("sys.cpu.user", ts, i * 2.0D, tags2);
             byte[] row2 = MetricAdapter.encodeRowKey(m2);
-            Key k2 = new Key(row2, tags2.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(
-                    ts, ""), new byte[0], ts);
+            Key k2 = new Key(row2, tags2.get(0).join().getBytes(StandardCharsets.UTF_8), new byte[0], new byte[0], ts);
             Value v2 = new Value(MetricAdapter.encodeValue(m2.getValue().getMeasure()));
             table.put(k2, v2);
         }
@@ -108,12 +108,8 @@ public class TimeSeriesGroupingIteratorTest extends IteratorTestBase {
         iter.init(source, settings.getOptions(), SCAN_IE);
         iter.seek(new Range(), EMPTY_COL_FAMS, true);
 
-        // this section changed when the key structure changed so that identical
-        // colFam values sorted consecutively within an given time period
         for (int i = 4; i < 100; i++) {
             checkNextResult(iter, new double[] { i - 4, i - 3, i - 2, i - 1, i });
-        }
-        for (int i = 4; i < 100; i++) {
             checkNextResult(iter, new double[] { (i - 4) * 2, (i - 3) * 2, (i - 2) * 2, (i - 1) * 2, i * 2 });
         }
         assertFalse(iter.hasTop());
@@ -132,21 +128,19 @@ public class TimeSeriesGroupingIteratorTest extends IteratorTestBase {
             ts += 1000;
             Metric m = new Metric("sys.cpu.user", ts, i * 1.0D, tags1);
             byte[] row = MetricAdapter.encodeRowKey(m);
-            Key k = new Key(row, tags1.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(ts,
-                    ""), new byte[0], ts);
+            Key k = new Key(row, tags1.get(0).join().getBytes(StandardCharsets.UTF_8), new byte[0], new byte[0], ts);
             Value v = new Value(MetricAdapter.encodeValue(m.getValue().getMeasure()));
             table.put(k, v);
             if (i < 50) {
                 // only populate this series 50 times
                 Metric m2 = new Metric("sys.cpu.user", ts, i * 2.0D, tags2);
                 byte[] row2 = MetricAdapter.encodeRowKey(m2);
-                Key k2 = new Key(row2, tags2.get(0).join().getBytes(StandardCharsets.UTF_8),
-                        MetricAdapter.encodeColQual(ts, ""), new byte[0], ts);
+                Key k2 = new Key(row2, tags2.get(0).join().getBytes(StandardCharsets.UTF_8), new byte[0], new byte[0],
+                        ts);
                 Value v2 = new Value(MetricAdapter.encodeValue(m2.getValue().getMeasure()));
                 table.put(k2, v2);
             }
         }
-
         SortedMapIterator source = new SortedMapIterator(table);
         TimeSeriesGroupingIterator iter = new TimeSeriesGroupingIterator();
         IteratorSetting settings = new IteratorSetting(100, TimeSeriesGroupingIterator.class);
@@ -154,15 +148,12 @@ public class TimeSeriesGroupingIteratorTest extends IteratorTestBase {
         iter.init(source, settings.getOptions(), SCAN_IE);
         iter.seek(new Range(), EMPTY_COL_FAMS, true);
 
-        // this section changed when the key structure changed so that identical
-        // colFam values sorted consecutively within an given time period
         for (int i = 4; i < 100; i++) {
             System.out.println(i);
             checkNextResult(iter, new double[] { i - 4, i - 3, i - 2, i - 1, i });
-        }
-        for (int i = 4; i < 50; i++) {
-            System.out.println(i);
-            checkNextResult(iter, new double[] { (i - 4) * 2, (i - 3) * 2, (i - 2) * 2, (i - 1) * 2, i * 2 });
+            if (i < 50) {
+                checkNextResult(iter, new double[] { (i - 4) * 2, (i - 3) * 2, (i - 2) * 2, (i - 1) * 2, i * 2 });
+            }
         }
         assertFalse(iter.hasTop());
     }
@@ -179,16 +170,15 @@ public class TimeSeriesGroupingIteratorTest extends IteratorTestBase {
             ts += 1000;
             Metric m = new Metric("sys.cpu.user", ts, i * 1.0D, tags1);
             byte[] row = MetricAdapter.encodeRowKey(m);
-            Key k = new Key(row, tags1.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(ts,
-                    ""), new byte[0], ts);
+            Key k = new Key(row, tags1.get(0).join().getBytes(StandardCharsets.UTF_8), new byte[0], new byte[0], ts);
             Value v = new Value(MetricAdapter.encodeValue(m.getValue().getMeasure()));
             table.put(k, v);
             if (i > 50) {
                 // only populate this series 50 times
                 Metric m2 = new Metric("sys.cpu.user", ts, i * 2.0D, tags2);
                 byte[] row2 = MetricAdapter.encodeRowKey(m2);
-                Key k2 = new Key(row2, tags2.get(0).join().getBytes(StandardCharsets.UTF_8),
-                        MetricAdapter.encodeColQual(ts, ""), new byte[0], ts);
+                Key k2 = new Key(row2, tags2.get(0).join().getBytes(StandardCharsets.UTF_8), new byte[0], new byte[0],
+                        ts);
                 Value v2 = new Value(MetricAdapter.encodeValue(m2.getValue().getMeasure()));
                 table.put(k2, v2);
             }
@@ -200,15 +190,12 @@ public class TimeSeriesGroupingIteratorTest extends IteratorTestBase {
         iter.init(source, settings.getOptions(), SCAN_IE);
         iter.seek(new Range(), EMPTY_COL_FAMS, true);
 
-        // this section changed when the key structure changed so that identical
-        // colFam values sorted consecutively within an given time period
         for (int i = 4; i < 100; i++) {
             checkNextResult(iter, new double[] { i - 4, i - 3, i - 2, i - 1, i });
+            if (i > 54) {
+                checkNextResult(iter, new double[] { (i - 4) * 2, (i - 3) * 2, (i - 2) * 2, (i - 1) * 2, i * 2 });
+            }
         }
-        for (int i = 55; i < 100; i++) {
-            checkNextResult(iter, new double[] { (i - 4) * 2, (i - 3) * 2, (i - 2) * 2, (i - 1) * 2, i * 2 });
-        }
-
         assertFalse(iter.hasTop());
 
     }
@@ -227,25 +214,22 @@ public class TimeSeriesGroupingIteratorTest extends IteratorTestBase {
             ts += 1000;
             Metric m = new Metric("sys.cpu.user", ts, i * 1.0D, tags1);
             byte[] row = MetricAdapter.encodeRowKey(m);
-            Key k = new Key(row, tags1.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(ts,
-                    ""), new byte[0], ts);
+            Key k = new Key(row, tags1.get(0).join().getBytes(StandardCharsets.UTF_8), new byte[0], new byte[0], ts);
             Value v = new Value(MetricAdapter.encodeValue(m.getValue().getMeasure()));
             table.put(k, v);
             // jitter the time on the second time series
             Metric m2 = new Metric("sys.cpu.user", ts + 50, i * 2.0D, tags2);
             byte[] row2 = MetricAdapter.encodeRowKey(m2);
-            Key k2 = new Key(row2, tags2.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(
-                    ts, ""), new byte[0], ts + 50);
+            Key k2 = new Key(row2, tags2.get(0).join().getBytes(StandardCharsets.UTF_8), new byte[0], new byte[0],
+                    ts + 50);
             Value v2 = new Value(MetricAdapter.encodeValue(m2.getValue().getMeasure()));
             table.put(k2, v2);
             Metric m3 = new Metric("sys.cpu.user", ts, i * 3.0D, tags3);
             byte[] row3 = MetricAdapter.encodeRowKey(m3);
-            Key k3 = new Key(row3, tags3.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(
-                    ts, ""), new byte[0], ts);
+            Key k3 = new Key(row3, tags3.get(0).join().getBytes(StandardCharsets.UTF_8), new byte[0], new byte[0], ts);
             Value v3 = new Value(MetricAdapter.encodeValue(m3.getValue().getMeasure()));
             table.put(k3, v3);
         }
-
         SortedMapIterator source = new SortedMapIterator(table);
         TimeSeriesGroupingIterator iter = new TimeSeriesGroupingIterator();
         IteratorSetting settings = new IteratorSetting(100, TimeSeriesGroupingIterator.class);
@@ -272,20 +256,23 @@ public class TimeSeriesGroupingIteratorTest extends IteratorTestBase {
         third.add(9D);
         third.add(12D);
 
-        // this section changed when the key structure changed so that identical
-        // colFam values sorted consecutively within an given time period
-        for (int i = 4; i < 100; i++) {
-            checkNextResult(iter, first);
-            shiftAndAdd(first, 1);
-        }
-        for (int i = 4; i < 100; i++) {
+        for (int i = 4; i < 196; i++) {
             System.out.println(i);
-            checkNextResult(iter, second);
-            shiftAndAdd(second, 2);
-        }
-        for (int i = 4; i < 100; i++) {
-            checkNextResult(iter, third);
-            shiftAndAdd(third, 3);
+            if (i == 4) {
+                checkNextResult(iter, first);
+                checkNextResult(iter, third);
+            } else if (i == 5) {
+                checkNextResult(iter, second);
+            } else if (i % 2 == 0) {
+                shiftAndAdd(first, 1);
+                checkNextResult(iter, first);
+
+                shiftAndAdd(third, 3);
+                checkNextResult(iter, third);
+            } else if (i % 2 != 0) {
+                shiftAndAdd(second, 2);
+                checkNextResult(iter, second);
+            }
         }
         assertFalse(iter.hasTop());
     }

server/src/test/java/timely/test/integration/DataStoreIT.java

@@ -49,7 +49,6 @@ public class DataStoreIT extends OneWaySSLBase {
             subQuery.setDownsample(Optional.of("1s-max"));
             request.addQuery(subQuery);
             List<QueryResponse> response = query("https://127.0.0.1:54322/api/query", request);
-
             assertEquals(1, response.size());
             Map<String, String> tags = response.get(0).getTags();
             assertEquals(1, tags.size());

server/src/test/java/timely/test/integration/http/HttpApiIT.java

@@ -572,11 +572,6 @@ public class HttpApiIT extends OneWaySSLBase {
         final Server s = new Server(conf);
         s.run();
         try {
-            System.out.println("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2 rack=r1");
-            System.out.println("sys.cpu.user " + (TEST_TIME + 1) + " 1.0 tag3=value3 rack=r2");
-            System.out.println("sys.cpu.idle " + (TEST_TIME + 2) + " 1.0 tag3=value3 tag4=value4 rack=r1");
-            System.out.println("sys.cpu.idle " + (TEST_TIME + 1000) + " 3.0 tag3=value3 tag4=value4 rack=r2");
-
             put("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2 rack=r1", "sys.cpu.user " + (TEST_TIME + 1)
                     + " 1.0 tag3=value3 rack=r2", "sys.cpu.idle " + (TEST_TIME + 2)
                     + " 1.0 tag3=value3 tag4=value4 rack=r1", "sys.cpu.idle " + (TEST_TIME + 1000)
@@ -595,11 +590,6 @@ public class HttpApiIT extends OneWaySSLBase {
             subQuery.setDownsample(Optional.of("1s-max"));
             request.addQuery(subQuery);
             List<QueryResponse> response = query("https://127.0.0.1:54322/api/query", request);
-
-            for (QueryResponse r : response) {
-                System.out.println(r.toString());
-            }
-
             assertEquals(2, response.size());
             QueryResponse response1 = response.get(0);
             Map<String, String> tags = response1.getTags();

server/src/test/java/timely/test/integration/websocket/WebSocketIT.java

@@ -346,15 +346,6 @@ public class WebSocketIT extends OneWaySSLBase {
             second = Metric.newBuilder().name("sys.cpu.user").value(TEST_TIME + 500, 7.0D).tag(new Tag("rack", "r2"))
                     .tag(new Tag("tag3", "value3")).build();
 
-            int x = 1;
-            for (String r : response) {
-                responses = JsonUtil.getObjectMapper().readValue(r, MetricResponses.class);
-                for (MetricResponse r1 : responses.getResponses()) {
-                    System.out.println(x + " -- " + r1.toString());
-                }
-                x++;
-            }
-
             Assert.assertEquals(1, response.size());
             responses = JsonUtil.getObjectMapper().readValue(response.get(0), MetricResponses.class);
             Assert.assertEquals(2, responses.size());