Release version 1.0.0

closes #166 : Gracefully handle tag values with delimiters and handle bad data

.gitignore

@@ -6,4 +6,5 @@ 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.6-SNAPSHOT</version>
+        <version>1.0.0</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.6-SNAPSHOT</version>
+        <version>1.0.0</version>
     </parent>
     <properties>
         <!-- PEP 440 compliant version gets supstituted into src/main/python/template/setup.py -->
-        <python.client.version>0.0.6.dev0</python.client.version>
+        <python.client.version>0.0.7.dev0</python.client.version>
     </properties>
     <dependencies>
         <dependency>
@@ -127,6 +127,7 @@
                     </execution>
                 </executions>
             </plugin>
+<!--
             <plugin>
                 <groupId>org.codehaus.mojo</groupId>
                 <artifactId>exec-maven-plugin</artifactId>
@@ -140,7 +141,7 @@
                         <configuration>
                             <executable>python</executable>
                             <workingDirectory>${project.basedir}/src/main/python</workingDirectory>
-                            <commandlineArgs>setup.py sdist --dist-dir ${project.build.directory}</commandlineArgs>
+                            <commandlineArgs>setup.py sdist - -dist-dir ${project.build.directory}</commandlineArgs>
                         </configuration>
                     </execution>
                     <execution>
@@ -157,6 +158,7 @@
                     </execution>
                 </executions>
             </plugin>
+-->
         </plugins>
     </build>
     <reporting>

client/src/main/java/timely/model/ObjectSizeOf.java

@@ -0,0 +1,200 @@
+package timely.model;
+
+import java.lang.reflect.Array;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.lang.reflect.Modifier;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.Stack;
+import java.util.logging.Logger;
+
+/**
+ * A simple interface that objects can implement to return the object size.
+ */
+public interface ObjectSizeOf {
+
+    /**
+     * The (approximate) size of the object
+     */
+    long sizeInBytes();
+
+    public static class ObjectInstance {
+
+        private Object o;
+
+        public ObjectInstance(Object _o) {
+            this.o = _o;
+        }
+
+        public Object getObject() {
+            return o;
+        }
+
+        @Override
+        public int hashCode() {
+            return System.identityHashCode(o);
+        }
+
+        @Override
+        public boolean equals(Object obj) {
+            if (!(obj instanceof ObjectInstance)) {
+                return false;
+            }
+            return ((ObjectInstance) obj).o == this.o;
+        }
+    }
+
+    public static class Sizer {
+
+        private static final Logger log = Logger.getLogger(Sizer.class.getCanonicalName());
+        public static final short OBJECT_OVERHEAD = 8;
+        public static final short ARRAY_OVERHEAD = 12;
+        public static final short REFERENCE = 4;
+        // The size of the basic Number constructs (and Boolean and Character)
+        // is 16: roundUp(8 + primitiveSize)
+        public static final short NUMBER_SIZE = 16;
+
+        /**
+         * Get the size of an object. Note that we want something relatively
+         * fast that gives us an order of magnitude here. The java
+         * Instrumentation agent mechanism is a little too costly for general
+         * use here. This will look for the ObjectSizeOf interface and if
+         * implemented on the object will use that. Otherwise it will do a
+         * simple navigation of the fields using reflection.
+         *
+         * @param o
+         * @return an approximation of the object size
+         */
+        public static long getObjectSize(Object o, boolean useSizeInBytesMethod, boolean debug) {
+            long estSize = getObjectSize(o, new HashSet<ObjectInstance>(), new Stack<ObjectInstance>(),
+                    useSizeInBytesMethod, debug);
+            return estSize;
+        }
+
+        public static long getObjectSize(Object o) {
+            long estSize = getObjectSize(o, new HashSet<ObjectInstance>(), new Stack<ObjectInstance>(), true, false);
+            return estSize;
+        }
+
+        public static long getObjectSize(Object o, Set<ObjectInstance> visited, Stack<ObjectInstance> stack,
+                boolean useSizeInBytesMethod, boolean debug) {
+            long totalSize = 0;
+            stack.add(new ObjectInstance(o));
+            while (!stack.isEmpty()) {
+                long size = 0;
+                ObjectInstance oi = stack.pop();
+                o = oi.getObject();
+                if (o != null && !visited.contains(oi)) {
+                    visited.add(oi);
+                    try {
+                        if (useSizeInBytesMethod) {
+                            try {
+                                if (o instanceof ObjectSizeOf) {
+                                    size = ((ObjectSizeOf) o).sizeInBytes();
+                                } else {
+                                    Method sizeInBytes = o.getClass().getMethod("sizeInBytes", (Class<?>[]) null);
+                                    size = (Long) sizeInBytes.invoke(o);
+                                }
+                            } catch (Throwable t) {
+                                // ok, lets do this the hard way...
+                            }
+                        }
+                        if (size == 0) {
+                            // the hard way...
+                            // do not include Class related objects or
+                            // reflection objects
+                            if ((o instanceof Class)
+                                    || (o instanceof ClassLoader)
+                                    || (o.getClass().getPackage() != null && o.getClass().getPackage().getName()
+                                            .startsWith("java.lang.reflect"))) {
+                                size = 0;
+                            } else if (o instanceof Number || o instanceof Boolean || o instanceof Character) {
+                                size = NUMBER_SIZE;
+                            } else {
+                                // lets do a simple sizing
+                                Class<?> c = o.getClass();
+                                if (c.isArray()) {
+                                    size = ARRAY_OVERHEAD;
+                                    int length = Array.getLength(o);
+                                    if (c.getComponentType().isPrimitive()) {
+                                        size += length * getPrimitiveObjectSize(c.getComponentType());
+                                    } else {
+                                        size += length * REFERENCE;
+                                        for (int i = 0; i < length; i++) {
+                                            Object element = Array.get(o, i);
+                                            if (element != null) {
+                                                stack.add(new ObjectInstance(element));
+                                            }
+                                        }
+                                    }
+                                } else {
+                                    size += OBJECT_OVERHEAD;
+                                    while (c != null) {
+                                        for (Field field : c.getDeclaredFields()) {
+                                            if (Modifier.isStatic(field.getModifiers())) {
+                                                continue;
+                                            }
+                                            if (field.getType().isPrimitive()) {
+                                                size += getPrimitiveObjectSize(field.getType());
+                                            } else {
+                                                size += REFERENCE;
+                                                boolean accessible = field.isAccessible();
+                                                field.setAccessible(true);
+                                                try {
+                                                    Object fieldObject = field.get(o);
+                                                    if (fieldObject != null) {
+                                                        stack.push(new ObjectInstance(fieldObject));
+                                                    }
+                                                } catch (Exception e) {
+                                                    // cannot get to field, so
+                                                    // ignore it in this size
+                                                    // calculation
+                                                    e.printStackTrace();
+                                                }
+                                                field.setAccessible(accessible);
+                                            }
+                                        }
+                                        c = c.getSuperclass();
+                                    }
+                                }
+                                size = roundUp(size);
+                            }
+                        }
+                    } catch (Throwable t) {
+                        log.warning("Unable to determine object size for " + o);
+                    }
+                    if (debug) {
+                        System.out.println(o.getClass().getCanonicalName() + "=" + size);
+                    }
+                }
+                totalSize += size;
+            }
+
+            return totalSize;
+        }
+
+        public static long roundUp(long size) {
+            long extra = size % 8;
+            if (extra > 0) {
+                size = size + 8 - extra;
+            }
+            return size;
+        }
+
+        public static short getPrimitiveObjectSize(Class<?> primitiveType) {
+            if (primitiveType.equals(int.class) || primitiveType.equals(float.class)) {
+                return 4;
+            } else if (primitiveType.equals(boolean.class) || primitiveType.equals(byte.class)) {
+                return 1;
+            } else if (primitiveType.equals(char.class) || primitiveType.equals(short.class)) {
+                return 2;
+            } else if (primitiveType.equals(long.class) || primitiveType.equals(double.class)) {
+                return 8;
+            } else { // if (primitiveType.equals(void.class)) {
+                return 0;
+            }
+        }
+    }
+
+}

client/src/main/java/timely/model/parse/TagListParser.java

@@ -1,6 +1,5 @@
 package timely.model.parse;
 
-import com.google.common.base.Splitter;
 import timely.model.Tag;
 
 import java.util.ArrayList;
@@ -15,13 +14,22 @@ import java.util.Map;
  */
 public class TagListParser implements Parser<List<Tag>>, Combiner<Collection<Tag>> {
 
-    private static final Splitter commaSplitter = Splitter.on(",").omitEmptyStrings();
     private static final TagParser tagParser = new TagParser();
 
     @Override
     public List<Tag> parse(String t) {
         ArrayList<Tag> tags = new ArrayList<>();
-        commaSplitter.splitToList(t).forEach(p -> tags.add(tagParser.parse(p)));
+        // only split on commas that are not preceded with a backslash
+        String[] parts = t.split("(?<!\\\\),");
+        for (String p : parts) {
+            if (!p.isEmpty()) {
+                Tag tag = tagParser.parse(p);
+                // unescape commas in the key and value of the tag
+                tag.setKey(tag.getKey().replaceAll("\\\\,", ","));
+                tag.setValue(tag.getValue().replaceAll("\\\\,", ","));
+                tags.add(tag);
+            }
+        }
         return tags;
     }
 

client/src/main/java/timely/model/parse/TagParser.java

@@ -12,7 +12,7 @@ import java.util.List;
  */
 public class TagParser implements Parser<Tag>, Combiner<Tag> {
 
-    private static final Splitter equalSplitter = Splitter.on("=").trimResults();
+    private static final Splitter equalSplitter = Splitter.on("=").limit(2).trimResults();
     private static final Joiner equalsJoiner = Joiner.on("=");
 
     @Override

client/src/test/java/timely/model/parse/MetricParserTest.java

@@ -0,0 +1,40 @@
+package timely.model.parse;
+
+import org.junit.Assert;
+import org.junit.Test;
+import timely.model.Metric;
+import timely.model.Tag;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class MetricParserTest {
+
+    @Test
+    public void testParseWithEscapedCharacters() {
+
+        MetricParser parser = new MetricParser();
+        Metric m = parser.parse("put mymetric 12341234 5.0 tag1=value1,value1 tag2=value2=value2");
+
+        Assert.assertEquals("mymetric", m.getName());
+        Assert.assertEquals(12341234, (long) m.getValue().getTimestamp());
+        Assert.assertEquals(5.0, (double) m.getValue().getMeasure(), 0);
+        List<Tag> expected = new ArrayList<>();
+        expected.add(new Tag("tag1", "value1,value1"));
+        expected.add(new Tag("tag2", "value2=value2"));
+        Assert.assertEquals(expected, m.getTags());
+    }
+
+    @Test
+    public void testParseMalformatted() {
+
+        MetricParser parser = new MetricParser();
+        try {
+            // parser should throw an exception
+            parser.parse("put mymetric 12341234 5.0 tag1 tag2=value2");
+            Assert.fail();
+        } catch (IllegalArgumentException e) {
+
+        }
+    }
+}

client/src/test/java/timely/model/parse/TagListParserTest.java

@@ -1,7 +1,8 @@
-package timely.model;
+package timely.model.parse;
 
 import org.junit.Assert;
 import org.junit.Test;
+import timely.model.Tag;
 import timely.model.parse.TagListParser;
 
 import java.util.ArrayList;
@@ -37,4 +38,15 @@ public class TagListParserTest {
         String combined = new TagListParser().combine(map);
         Assert.assertEquals("tag1=value1,tag2=value2", combined);
     }
+
+    @Test
+    public void testParseTagsWithCommas() {
+
+        try {
+            String s = "tag1=value1,tag2=3.4.3_(default\\,_Date\\,_Time)_";
+            new TagListParser().parse(s);
+        } catch (Exception e) {
+            Assert.fail(e.getMessage());
+        }
+    }
 }

collectd-nsq-plugin/pom.xml

@@ -4,7 +4,7 @@
     <parent>
         <groupId>timely</groupId>
         <artifactId>timely</artifactId>
-        <version>0.0.6-SNAPSHOT</version>
+        <version>1.0.0</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.6-SNAPSHOT</version>
+        <version>1.0.0</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.6-SNAPSHOT</version>
+        <version>1.0.0</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.6-SNAPSHOT</version>
+    <version>1.0.0</version>
   </parent>
   <artifactId>timely-common</artifactId>
   <dependencies>

grafana/pom.xml

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

pom.xml

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

server/conf/timely-docker.yml

@@ -15,6 +15,7 @@ timely:
       bufferSize: 52428800
     scan:
       threads: 4
+      max-downsample-memory: 10000000
   security:
     session-max-age: 86400
     allow-anonymous-access: true

server/conf/timely-standalone.yml

@@ -15,6 +15,7 @@ timely:
       bufferSize: 52428800
     scan:
       threads: 4
+      max-downsample-memory: 10000000
   security:
     session-max-age: 86400
     allow-anonymous-access: true

server/conf/timely.yml

@@ -15,6 +15,7 @@ timely:
       bufferSize: 52428800
     scan:
       threads: 4
+      max-downsample-memory: 10000000
   security:
     session-max-age: 86400
     allow-anonymous-access: false

server/pom.xml

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

server/src/main/java/timely/Configuration.java

@@ -218,6 +218,7 @@ public class Configuration {
     public class Scan {
 
         private int threads = 4;
+        private long maxDownsampleMemory = -1;
 
         public int getThreads() {
             return threads;
@@ -227,6 +228,15 @@ public class Configuration {
             this.threads = threads;
             return Configuration.this;
         }
+
+        public long getMaxDownsampleMemory() {
+            return maxDownsampleMemory;
+        }
+
+        public Configuration setMaxDownsampleMemory(long maxDownsampleMemory) {
+            this.maxDownsampleMemory = maxDownsampleMemory;
+            return Configuration.this;
+        }
     }
 
     public class Security {

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

@@ -9,12 +9,15 @@ 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.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
@@ -28,16 +31,43 @@ 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;
+    }
+
+    private static List<Tag> escapeDelimiters(List<Tag> tags) {
+        List<Tag> newTags = new ArrayList<>();
+        for (Tag t : tags) {
+            // escape all commas and equals in the tag key and value since they
+            // are used as a delimiter
+            String k = t.getKey();
+            String v = t.getValue();
+            k = k.replaceAll("=", "\\=");
+            k = k.replaceAll(",", "\\,");
+            v = v.replaceAll("=", "\\=");
+            v = v.replaceAll(",", "\\,");
+            newTags.add(new Tag(k, v));
+        }
+        return newTags;
+    }
+
     public static Mutation toMutation(Metric metric) {
         final Mutation mutation = new Mutation(encodeRowKey(metric));
-
         List<Tag> tags = metric.getTags();
+        tags = escapeDelimiters(tags);
         Collections.sort(tags);
 
         for (final Tag entry : tags) {
@@ -51,8 +81,9 @@ public class MetricAdapter {
                     .map(Tag::join)
                     .collect(Collectors.joining(","));
             // @formatter:on
-
-            mutation.put(cf, cq, extractVisibility(tags), metric.getValue().getTimestamp(), extractValue(metric));
+            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));
         }
         return mutation;
     }
@@ -87,10 +118,11 @@ public class MetricAdapter {
         // @formatter:off
         Metric.Builder builder = Metric.newBuilder()
                 .name(row.getFirst())
-                .value(row.getSecond(), ByteBuffer.wrap(v.get()).getDouble())
+                .value(k.getTimestamp(), ByteBuffer.wrap(v.get()).getDouble())
                 .tag(tagParser.parse(k.getColumnFamily().toString()));
         // @formatter:on
-        tagListParser.parse(k.getColumnQualifier().toString()).forEach(builder::tag);
+        ComparablePair<Long, String> cq = colQualCoder.decode(k.getColumnQualifier().getBytes());
+        tagListParser.parse(cq.getSecond()).forEach(builder::tag);
         if (includeVizTag && k.getColumnVisibility().getLength() > 0) {
             tagListParser.parse("viz=" + k.getColumnVisibility().toString()).forEach(builder::tag);
         }
@@ -106,7 +138,16 @@ public class MetricAdapter {
     }
 
     public static byte[] encodeRowKey(Metric metric) {
-        return encodeRowKey(metric.getName(), metric.getValue().getTimestamp());
+        // 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);
     }
 
     public static Pair<String, Long> decodeRowKey(Key k) {

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

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

server/src/main/java/timely/api/response/timeseries/QueryResponse.java

@@ -2,9 +2,9 @@ package timely.api.response.timeseries;
 
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.TreeMap;
 
 import org.apache.commons.lang3.builder.ToStringBuilder;
 
@@ -13,7 +13,7 @@ public class QueryResponse {
     private String metric;
     private Map<String, String> tags = new HashMap<>();
     private List<String> aggregatedTags = new ArrayList<>();
-    private Map<String, Object> dps = new TreeMap<>();
+    private Map<String, Object> dps = new LinkedHashMap<>();
 
     public String getMetric() {
         return metric;

server/src/main/java/timely/sample/Aggregation.java

@@ -3,6 +3,7 @@ package timely.sample;
 import com.google.common.base.Preconditions;
 import org.apache.commons.lang3.builder.ToStringBuilder;
 import timely.api.request.timeseries.QueryRequest;
+import timely.model.ObjectSizeOf;
 
 import java.io.IOException;
 import java.io.ObjectInputStream;
@@ -140,4 +141,7 @@ public class Aggregation implements Iterable<Sample>, Serializable {
         return result;
     }
 
+    public int getNumBuckets() {
+        return buckets.size();
+    }
 }

server/src/main/java/timely/sample/Downsample.java

@@ -3,6 +3,7 @@ package timely.sample;
 import com.google.common.base.Preconditions;
 import org.apache.commons.lang3.builder.ToStringBuilder;
 import timely.api.request.timeseries.QueryRequest.RateOption;
+import timely.model.ObjectSizeOf;
 
 import java.io.Serializable;
 import java.util.ArrayList;
@@ -64,5 +65,4 @@ public class Downsample extends Aggregation implements Iterable<Sample>, Seriali
         }
         return result;
     }
-
 }

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

@@ -5,10 +5,7 @@ import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
+import java.util.*;
 
 import org.apache.accumulo.core.client.IteratorSetting;
 import org.apache.accumulo.core.data.Key;
@@ -17,8 +14,10 @@ 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 timely.adapter.accumulo.MetricAdapter;
 import timely.model.Metric;
+import timely.model.ObjectSizeOf;
 import timely.model.Tag;
 import timely.api.response.TimelyException;
 import timely.sample.Aggregator;
@@ -27,23 +26,36 @@ 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";
+    private static final String MAX_DOWNSAMPLE_MEMORY = "downsample.maxDownsampleMemory";
     private static final String AGGCLASS = "downsample.aggclass";
 
     private DownsampleFactory factory;
     private final Map<Set<Tag>, Downsample> value = new HashMap<>();
+    private long start;
+    private long end;
+    private long period;
     private Key last;
+    private DownsampleMemoryEstimator memoryEstimator = null;
 
     @SuppressWarnings("unchecked")
     @Override
     public void init(SortedKeyValueIterator<Key, Value> source, Map<String, String> options, IteratorEnvironment env)
             throws IOException {
         super.init(source, options, env);
-        long start = Long.parseLong(options.get(START));
-        long end = Long.parseLong(options.get(END));
-        long period = Long.parseLong(options.get(PERIOD));
+        start = Long.parseLong(options.get(START));
+        end = Long.parseLong(options.get(END));
+        period = Long.parseLong(options.get(PERIOD));
+        // default = 100 MB
+        long maxDownsampleMemory = -1;
+        if (options.containsKey(MAX_DOWNSAMPLE_MEMORY)) {
+            maxDownsampleMemory = Long.parseLong(options.get(MAX_DOWNSAMPLE_MEMORY));
+        }
+        memoryEstimator = new DownsampleMemoryEstimator(maxDownsampleMemory, start, period);
+
         String aggClassname = options.get(AGGCLASS);
         Class<?> aggClass;
         try {
@@ -56,26 +68,54 @@ public class DownsampleIterator extends WrappingIterator {
 
     @Override
     public boolean hasTop() {
-        while (super.hasTop()) {
-            last = super.getTopKey();
-            Metric metric = MetricAdapter.parse(super.getTopKey(), super.getTopValue());
-            Set<Tag> tags = new HashSet<Tag>(metric.getTags());
-            Downsample sample = value.get(tags);
-            if (sample == null) {
+
+        if (super.hasTop()) {
+            Key topKey = super.getTopKey();
+            Value topValue = super.getTopValue();
+            Metric metric = null;
+            try {
+                metric = MetricAdapter.parse(topKey, topValue);
+                LOG.trace("entering hasTop() for metric=" + metric.toString());
+            } catch (Exception e) {
+                LOG.error("Error: {} parsing metric at key: {}", e.getMessage(), topKey.toString());
+            }
+            while (super.hasTop()) {
+                topKey = super.getTopKey();
+                topValue = super.getTopValue();
                 try {
-                    value.put(tags, sample = factory.create());
-                } catch (TimelyException e) {
-                    throw new RuntimeException(e);
+                    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;
+
+                    Set<Tag> tags = new HashSet<Tag>(metric.getTags());
+                    Downsample sample = value.get(tags);
+                    if (sample == null) {
+                        try {
+                            value.put(tags, sample = factory.create());
+                        } catch (TimelyException e) {
+                            throw new RuntimeException(e);
+                        }
+                    }
+                    sample.add(metric.getValue().getTimestamp(), metric.getValue().getMeasure());
+                } catch (Exception e) {
+                    LOG.error("Error: {} parsing metric at key: {}", e.getMessage(), topKey.toString());
+                }
+                try {
+                    super.next();
+                } catch (IOException e) {
+                    throw new RuntimeException("Downstream next() failed", e);
                 }
             }
-            sample.add(metric.getValue().getTimestamp(), metric.getValue().getMeasure());
-            try {
-                super.next();
-            } catch (IOException e) {
-                throw new RuntimeException("Downstream next() failed", e);
-            }
+            LOG.trace("exiting hasTop() for metric=" + metric.toString());
+            return last != null;
+        } else {
+            return false;
         }
-        return last != null;
     }
 
     @Override
@@ -89,6 +129,9 @@ public class DownsampleIterator extends WrappingIterator {
                 ObjectOutputStream out = new ObjectOutputStream(bos)) {
             out.writeObject(value);
             out.flush();
+            // empty for next batch of downsamples
+            value.clear();
+            memoryEstimator.reset();
             return new Value(bos.toByteArray());
         } catch (IOException e) {
             throw new RuntimeException(e);
@@ -97,13 +140,17 @@ public class DownsampleIterator extends WrappingIterator {
 
     @Override
     public void next() throws IOException {
-        last = null;
+        if (!super.hasTop()) {
+            last = null;
+        }
     }
 
-    public static void setDownsampleOptions(IteratorSetting is, long start, long end, long period, String classname) {
+    public static void setDownsampleOptions(IteratorSetting is, long start, long end, long period,
+            long maxDownsampleMemory, String classname) {
         is.addOption(START, "" + start);
         is.addOption(END, "" + end);
         is.addOption(PERIOD, "" + period);
+        is.addOption(MAX_DOWNSAMPLE_MEMORY, "" + maxDownsampleMemory);
         is.addOption(AGGCLASS, classname);
     }
 

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

@@ -0,0 +1,101 @@
+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,9 +2,12 @@ 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;
@@ -13,6 +16,7 @@ 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 {
 
@@ -26,6 +30,12 @@ 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,6 +5,7 @@ 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.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -45,11 +46,7 @@ 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.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.data.*;
 import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
 import org.apache.accumulo.core.iterators.user.RegExFilter;
 import org.apache.accumulo.core.security.Authorizations;
@@ -125,6 +122,7 @@ public class DataStoreImpl implements DataStore {
     private final InternalMetrics internalMetrics = new InternalMetrics();
     private final Timer internalMetricsTimer = new Timer(true);
     private final int scannerThreads;
+    private final long maxDownsampleMemory;
     private final BatchWriterConfig bwConfig;
     private final List<BatchWriter> writers = new ArrayList<>();
     private final ThreadLocal<BatchWriter> metaWriter = new ThreadLocal<>();
@@ -149,6 +147,7 @@ public class DataStoreImpl implements DataStore {
             bwConfig.setMaxMemory(getMemoryInBytes(accumuloConf.getWrite().getBufferSize()) / numWriteThreads);
             bwConfig.setMaxWriteThreads(accumuloConf.getWrite().getThreads());
             scannerThreads = accumuloConf.getScan().getThreads();
+            maxDownsampleMemory = accumuloConf.getScan().getMaxDownsampleMemory();
             anonAccessAllowed = conf.getSecurity().isAllowAnonymousAccess();
 
             metricsTable = conf.getMetricsTable();
@@ -312,7 +311,6 @@ public class DataStoreImpl implements DataStore {
             metaCache.addAll(toCache);
         }
         try {
-
             batchWriter.get().addMutation(MetricAdapter.toMutation(metric));
             internalMetrics.incrementMetricKeysInserted(metric.getTags().size());
         } catch (MutationsRejectedException e) {
@@ -464,13 +462,20 @@ 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 {
@@ -484,10 +489,12 @@ public class DataStoreImpl implements DataStore {
                     long endOfLastPeriod = (endDistanceFromDownSample > 0 ? endTs + downsample
                             - endDistanceFromDownSample : endTs);
 
-                    setQueryRange(scanner, metric, startOfFirstPeriod, endOfLastPeriod);
-                    List<String> tagOrder = prioritizeTags(query);
+                    List<String> tagOrder = prioritizeTags(query.getMetric(), query.getTags());
                     Map<String, String> orderedTags = orderTags(tagOrder, query.getTags());
-                    setQueryColumns(scanner, metric, orderedTags);
+                    Set<Tag> colFamValues = getColumnFamilies(metric, orderedTags);
+                    List<Range> ranges = getQueryRanges(metric, startOfFirstPeriod, endOfLastPeriod, colFamValues);
+                    scanner.setRanges(ranges);
+                    setQueryColumns(scanner, metric, orderedTags, colFamValues);
 
                     if (query.isRate()) {
                         LOG.trace("Adding rate iterator");
@@ -506,7 +513,7 @@ public class DataStoreImpl implements DataStore {
                         LOG.trace("Downsample Aggregator type {}", daggClass.getSimpleName());
                         IteratorSetting is = new IteratorSetting(500, DownsampleIterator.class);
                         DownsampleIterator.setDownsampleOptions(is, startOfFirstPeriod, endOfLastPeriod, downsample,
-                                daggClass.getName());
+                                maxDownsampleMemory, daggClass.getName());
                         scanner.addScanIterator(is);
                     }
 
@@ -527,6 +534,7 @@ 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())) {
@@ -546,10 +554,12 @@ 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: {}", (System.currentTimeMillis() - now));
+            LOG.debug("Query time: {} duration: {} metrics: {} results: {}", (System.currentTimeMillis() - now),
+                    (endTs - startTs), metricList.toString(), numResults);
             internalMetrics.addQueryResponse(result.size(), (System.currentTimeMillis() - now));
             return result;
         } catch (ClassNotFoundException | IOException | TableNotFoundException ex) {
@@ -577,10 +587,9 @@ public class DataStoreImpl implements DataStore {
      * @param query
      * @return ordered list of most specific to least specific tags in the query
      */
-    private List<String> prioritizeTags(SubQuery query) {
+    private List<String> prioritizeTags(String metric, Map<String, String> tags) {
         // trivial cases
-        Map<String, String> tags = query.getTags();
-        if (tags.isEmpty()) {
+        if (tags == null || tags.isEmpty()) {
             return Collections.emptyList();
         }
         if (tags.size() == 1) {
@@ -588,7 +597,6 @@ 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()) {
@@ -644,8 +652,9 @@ public class DataStoreImpl implements DataStore {
         return !REGEX_TEST.matcher(value).matches();
     }
 
-    private void setQueryColumns(ScannerBase scanner, String metric, Map<String, String> tags)
-            throws TableNotFoundException, TimelyException {
+    public Set<Tag> getColumnFamilies(String metric, Map<String, String> requestedTags) throws TableNotFoundException {
+
+        Map<String, String> tags = (requestedTags == null) ? new LinkedHashMap<>() : requestedTags;
         LOG.trace("Looking for requested tags: {}", tags);
         Scanner meta = connector.createScanner(metaTable, Authorizations.EMPTY);
         Text start = new Text(Meta.VALUE_PREFIX + metric);
@@ -708,24 +717,35 @@ 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.
-        Set<Tag> concrete = expandTagValues(first, knownKeyValues);
-        if (concrete.size() == 0) {
+        return expandTagValues(first, knownKeyValues);
+    }
+
+    private void setQueryColumns(ScannerBase scanner, String metric, Map<String, String> tags, Set<Tag> colFamValues)
+            throws TimelyException {
+
+        if (colFamValues.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: {}", concrete);
-        for (Tag tag : concrete) {
+        LOG.trace("Found matching tags: {}", colFamValues);
+        for (Tag tag : colFamValues) {
             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()) {
-            Entry<String, String> tag = tagIter.next();
+            tag = tagIter.next();
             LOG.trace("Adding regex filter for tag {}", tag);
             StringBuffer pattern = new StringBuffer();
-            pattern.append("(^|.*,)");
+            pattern.append("(^.*\\x00|.*,)");
             pattern.append(tag.getKey());
             pattern.append("=");
             pattern.append(tag.getValue());
@@ -768,14 +788,55 @@ public class DataStoreImpl implements DataStore {
         return result;
     }
 
-    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));
+    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 Class<? extends Aggregator> getAggregator(SubQuery query) {
@@ -836,7 +897,7 @@ public class DataStoreImpl implements DataStore {
         }
     }
 
-    private long getAgeOffForMetric(String metricName) {
+    public long getAgeOffForMetric(String metricName) {
         String age = this.ageOff.get(MetricAgeOffIterator.AGE_OFF_PREFIX + metricName);
         if (null == age) {
             return this.defaultAgeOff;
@@ -866,23 +927,13 @@ public class DataStoreImpl implements DataStore {
             if (null == metric) {
                 throw new IllegalArgumentException("metric name must be specified");
             }
-            if (0 == startTime) {
-                startTime = (System.currentTimeMillis() - getAgeOffForMetric(metric) - 1000);
-                LOG.debug("Overriding zero start time to {} due to age off configuration", startTime);
+            if (tags == null) {
+                tags = new LinkedHashMap<>();
             }
-            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);
+            List<String> tagOrder = prioritizeTags(metric, tags);
+            Map<String, String> orderedTags = orderTags(tagOrder, tags);
+            Set<Tag> colFamValues = getColumnFamilies(metric, orderedTags);
+            setQueryColumns(s, metric, orderedTags, colFamValues);
             s.setBatchSize(scannerBatchSize);
             s.setReadaheadThreshold(scannerReadAhead);
             return s;

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

@@ -231,29 +231,33 @@ public class TimeSeriesGroupingIterator extends WrappingIterator {
         while (super.hasTop()) {
             Key k = super.getTopKey();
 
-            Long newTime = MetricAdapter.decodeRowKey(k).getSecond();
+            Long newTime = k.getTimestamp();
             if (time == null) {
                 time = newTime;
             }
             if (time.equals(newTime)) {
 
-                Metric m = MetricAdapter.parse(k, super.getTopValue());
-                timely.model.Value v = m.getValue();
-                m.setValue(null);
+                try {
+                    Metric m = MetricAdapter.parse(k, super.getTopValue());
+                    timely.model.Value v = m.getValue();
+                    m.setValue(null);
 
-                TimeSeries values = series.get(m);
-                if (null == values) {
-                    LOG.trace("Creating new time series {}", m);
-                    values = new TimeSeries(this, filters.length);
+                    TimeSeries values = series.get(m);
+                    if (null == values) {
+                        LOG.trace("Creating new time series {}", m);
+                        values = new TimeSeries(this, filters.length);
+                    }
+
+                    LOG.trace("Adding value {} to series {}", v.getMeasure(), m);
+                    values.add(k, v.getMeasure());
+
+                    // always re-put the metric back into the TimeSeriesGroup as
+                    // it
+                    // maintains a list of prepared answers
+                    series.put(m, values);
+                } catch (Exception e) {
+                    LOG.error("Error: {} parsing metric at key: {}", e.getMessage(), k.toString());
                 }
-
-                LOG.trace("Adding value {} to series {}", v.getMeasure(), m);
-                values.add(k, v.getMeasure());
-
-                // always re-put the metric back into the TimeSeriesGroup as it
-                // maintains a list of prepared answers
-                series.put(m, values);
-
                 super.next();
             } else {
                 break;

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

@@ -2,16 +2,20 @@ 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;
@@ -25,6 +29,7 @@ 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;
 
@@ -45,11 +50,16 @@ public class MetricScanner extends Thread implements UncaughtExceptionHandler {
     private final int lag;
     private final String subscriptionId;
     private final String metric;
-    private final long endTime;
+    private long startTime;
+    private 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,
@@ -57,13 +67,34 @@ 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 = store.createScannerForMetric(sessionId, metric, tags, startTime, endTime, lag, scannerBatchSize,
-                scannerReadAhead);
+        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.iter = scanner.iterator();
         this.delay = delay;
         this.subscriptionId = subscriptionId;
@@ -86,6 +117,7 @@ public class MetricScanner extends Thread implements UncaughtExceptionHandler {
     }
 
     public synchronized void flush() {
+        LOG.info("Flush called");
         synchronized (responses) {
             if (responses.size() > 0) {
                 try {
@@ -103,48 +135,69 @@ public class MetricScanner extends Thread implements UncaughtExceptionHandler {
     public void run() {
         Metric m = null;
         try {
-            while (!closed) {
+            boolean done = false;
+            while (!done && !closed) {
 
                 if (this.iter.hasNext()) {
                     Entry<Key, Value> e = this.iter.next();
-                    m = MetricAdapter.parse(e.getKey(), e.getValue(), true);
-                    if (responses.size() >= this.subscriptionBatchSize) {
-                        flush();
+                    try {
+                        m = MetricAdapter.parse(e.getKey(), e.getValue(), true);
+                        if (responses.size() >= this.subscriptionBatchSize) {
+                            flush();
+                        }
+                        this.responses.addResponse(MetricResponse.fromMetric(m, this.subscriptionId));
+                    } catch (Exception e1) {
+                        LOG.error("Error {} parsing metric at key: {}", e1.getMessage(), e.getKey().toString());
                     }
-                    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);
-                        this.scanner.setRange(new Range(prevRange.getStartKey().getRow(), prevRange
-                                .isStartKeyInclusive(), endRow, false));
-                        this.iter = this.scanner.iterator();
+                        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();
+                        }
                     } 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);
-                        this.scanner.setRange(new Range(new Text(MetricAdapter.encodeRowKey(m)), false, endRow,
-                                prevRange.isEndKeyInclusive()));
-                        this.iter = this.scanner.iterator();
+                        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;
+                        }
                     }
                 } 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 (Exception e) {
+        } catch (Throwable e) {
             LOG.error("Error in metric scanner, closing.", e);
         } finally {
             close();

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

@@ -15,6 +15,8 @@ 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() {
@@ -25,7 +27,8 @@ public class TimelyMetricsFormatter extends DefaultFormatter {
         b.append(" ");
         b.append(e.getKey().getColumnFamily().toString());
         b.append(":");
-        b.append(e.getKey().getColumnQualifier().toString());
+        ComparablePair<Long, String> cq = colQualCoder.decode(e.getKey().getColumnQualifier().copyBytes());
+        b.append(cq.getFirst()).append("\\x00").append(cq.getSecond());
         b.append(" ");
         b.append(e.getKey().getColumnVisibility().toString());
         if (super.isDoTimestamps()) {

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

@@ -42,28 +42,33 @@ 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", ts));
+        byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", MetricAdapter.roundTimestampToLastHour(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.assertTrue(new String(up.getColumnFamily()).equals("tag1=value1"));
-        Assert.assertTrue(new String(up.getColumnQualifier()).equals("tag2=value2,tag3=value3"));
+        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.assertEquals(ts, up.getTimestamp());
-        Assert.assertTrue(new String(up.getColumnVisibility()).equals(""));
+        Assert.assertEquals("", new String(new String(up.getColumnVisibility())));
         Assert.assertArrayEquals(value, up.getValue());
         ColumnUpdate up2 = mut.getUpdates().get(1);
-        Assert.assertTrue(new String(up2.getColumnFamily()).equals("tag2=value2"));
-        Assert.assertTrue(new String(up2.getColumnQualifier()).equals("tag1=value1,tag3=value3"));
+        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.assertEquals(ts, up2.getTimestamp());
-        Assert.assertTrue(new String(up2.getColumnVisibility()).equals(""));
+        Assert.assertEquals("", new String(up2.getColumnVisibility()));
         Assert.assertArrayEquals(value, up.getValue());
         ColumnUpdate up3 = mut.getUpdates().get(2);
-        Assert.assertTrue(new String(up3.getColumnFamily()).equals("tag3=value3"));
-        Assert.assertTrue(new String(up3.getColumnQualifier()).equals("tag1=value1,tag2=value2"));
+        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.assertEquals(ts, up3.getTimestamp());
-        Assert.assertTrue(new String(up3.getColumnVisibility()).equals(""));
+        Assert.assertEquals("", new String(up3.getColumnVisibility()));
         Assert.assertArrayEquals(value, up.getValue());
     }
 
@@ -78,14 +83,16 @@ 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", ts));
+        byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", MetricAdapter.roundTimestampToLastHour(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()));
-        Assert.assertEquals("", new String(up.getColumnQualifier()));
+        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(ts, up.getTimestamp());
         Assert.assertEquals("(a&b)|(c&d)", new String(up.getColumnVisibility()));
         Assert.assertArrayEquals(value, up.getValue());
@@ -110,8 +117,9 @@ 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);
-        Key k = new Key(row, "tag1=value1".getBytes(), "tag2=value2,tag3=value3".getBytes(), "(a&b)|(c&d)".getBytes(),
-                1000);
+        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);
         Value v = new Value(value);
         Metric m = MetricAdapter.parse(k, v);
         Assert.assertEquals("sys.cpu.user", m.getName());
@@ -124,14 +132,36 @@ public class MetricAdapterTest {
         Assert.assertEquals(2.0D, m.getValue().getMeasure(), 0.0D);
     }
 
+    @Test
+    public void testParseWithEscapes() throws Exception {
+        PairLexicoder<String, Long> rowCoder = new PairLexicoder<>(new StringLexicoder(), new LongLexicoder());
+        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=test\\,value2,tag3=val\\=ue3")), "(a&b)|(c&d)".getBytes(), 1000);
+        Value v = new Value(value);
+        Metric m = MetricAdapter.parse(k, v);
+        Assert.assertEquals("sys.cpu.user", m.getName());
+        List<Tag> tags = new ArrayList<>();
+        tags.add(new Tag("tag1=value1"));
+        tags.add(new Tag("tag2=test,value2"));
+        tags.add(new Tag("tag3=val\\=ue3"));
+        Assert.assertEquals(tags, m.getTags());
+        Assert.assertEquals(new Long(1000), m.getValue().getTimestamp());
+        Assert.assertEquals(2.0D, m.getValue().getMeasure(), 0.0D);
+    }
+
     @Test
     public void testParseWithViz() throws Exception {
         PairLexicoder<String, Long> rowCoder = new PairLexicoder<>(new StringLexicoder(), new LongLexicoder());
         byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", 1000L));
         byte[] value = new byte[Double.BYTES];
         ByteBuffer.wrap(value).putDouble(2.0D);
-        Key k = new Key(row, "tag1=value1".getBytes(), "tag2=value2,tag3=value3".getBytes(), "(a&b)|(c&d)".getBytes(),
-                1000);
+        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);
         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

@@ -4,13 +4,8 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.Map.Entry;
-import java.util.Set;
-import java.util.SortedMap;
-import java.util.TreeMap;
 
 import org.apache.accumulo.core.client.IteratorSetting;
 import org.apache.accumulo.core.data.ColumnUpdate;
@@ -26,8 +21,10 @@ import org.junit.Test;
 import timely.Configuration;
 import timely.adapter.accumulo.MetricAdapter;
 import timely.model.Metric;
+import timely.model.ObjectSizeOf;
 import timely.model.Tag;
 import timely.auth.VisibilityCache;
+import timely.model.parse.TagListParser;
 import timely.sample.Downsample;
 import timely.sample.Sample;
 import timely.sample.aggregators.Avg;
@@ -57,6 +54,7 @@ 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()));
             }
         }
@@ -85,7 +83,7 @@ public class DownsampleIteratorTest {
     public void simpleGetOneSample() throws Exception {
         // check that data gets pulled out
         DownsampleIterator iter = new DownsampleIterator();
-        Map<Set<Tag>, Downsample> samples = runQuery(iter, testData1, 100);
+        Map<Set<Tag>, Downsample> samples = runQuery(iter, testData1, 100, -1);
         assertEquals(1, samples.size());
         for (Entry<Set<Tag>, Downsample> entry : samples.entrySet()) {
             Set<Tag> tags = entry.getKey();
@@ -104,7 +102,7 @@ public class DownsampleIteratorTest {
     @Test
     public void simpleGetTwoSamples() throws Exception {
         DownsampleIterator iter = new DownsampleIterator();
-        Map<Set<Tag>, Downsample> samples = runQuery(iter, testData2, 100);
+        Map<Set<Tag>, Downsample> samples = runQuery(iter, testData2, 100, -1);
         assertEquals(2, samples.size());
         for (Tag tag : new Tag[] { new Tag("host", "host1"), new Tag("host", "host2") }) {
             Downsample dsample = samples.get(Collections.singleton(tag));
@@ -128,7 +126,7 @@ public class DownsampleIteratorTest {
     @Test
     public void simpleTestDownsampling() throws Exception {
         DownsampleIterator iter = new DownsampleIterator();
-        Map<Set<Tag>, Downsample> samples = runQuery(iter, testData2, 200);
+        Map<Set<Tag>, Downsample> samples = runQuery(iter, testData2, 200, -1);
         assertEquals(2, samples.size());
         for (Tag tag : new Tag[] { new Tag("host", "host1"), new Tag("host", "host2") }) {
             Downsample dsample = samples.get(Collections.singleton(tag));
@@ -149,17 +147,142 @@ public class DownsampleIteratorTest {
         }
     }
 
+    static public class SampleObject implements ObjectSizeOf {
+
+        private long sizeInBytes = 0;
+
+        public void setSizeInBytes(long sizeInBytes) {
+            this.sizeInBytes = sizeInBytes;
+        }
+
+        @Override
+        public long sizeInBytes() {
+            return sizeInBytes;
+        }
+    }
+
+    @Test
+    public void memoryEstimatorTestSmallObjects() {
+        long maxMemory = 1000;
+        long start = System.currentTimeMillis();
+        long period = 500l;
+        long sizeOfObjects = 20;
+        SampleObject o = new SampleObject();
+        DownsampleMemoryEstimator memoryEstimator = new DownsampleMemoryEstimator(maxMemory, start, period);
+        boolean shouldReturn = false;
+        for (long x = 100; x <= 5000; x += 100) {
+            long timestamp = start + x;
+            o.setSizeInBytes(o.sizeInBytes() + sizeOfObjects);
+            shouldReturn = memoryEstimator.shouldReturnBasedOnMemoryUsage(timestamp, o);
+            if (memoryEstimator.isNewBucket()) {
+                long memoryPercentageUsedCalculated = Math.round((double) o.sizeInBytes() / maxMemory * 100);
+                long memoryPercentageUsedEstimate = Math.round(memoryEstimator.getMemoryUsedPercentage());
+                long percentError = Math.round(Math.abs(memoryPercentageUsedCalculated - memoryPercentageUsedEstimate)
+                        / memoryPercentageUsedCalculated * 100);
+                assertTrue(percentError == 0);
+            }
+
+            if (shouldReturn) {
+                o.setSizeInBytes(0);
+                memoryEstimator.reset();
+            }
+        }
+        assertTrue(shouldReturn);
+    }
+
+    @Test
+    public void memoryEstimatorTestLargeObjects() {
+        long maxMemory = 10000;
+        long start = System.currentTimeMillis();
+        long period = 500l;
+        long sizeOfObjects = 200;
+        SampleObject o = new SampleObject();
+        DownsampleMemoryEstimator memoryEstimator = new DownsampleMemoryEstimator(maxMemory, start, period);
+        boolean shouldReturn = false;
+        for (long x = 100; x <= 5000; x += 100) {
+            long timestamp = start + x;
+            o.setSizeInBytes(o.sizeInBytes() + sizeOfObjects);
+            shouldReturn = memoryEstimator.shouldReturnBasedOnMemoryUsage(timestamp, o);
+            if (memoryEstimator.isNewBucket()) {
+                long memoryPercentageUsedCalculated = Math.round((double) o.sizeInBytes() / maxMemory * 100);
+                long memoryPercentageUsedEstimate = Math.round(memoryEstimator.getMemoryUsedPercentage());
+                long percentError = Math.round(Math.abs(memoryPercentageUsedCalculated - memoryPercentageUsedEstimate)
+                        / memoryPercentageUsedCalculated * 100);
+                assertTrue(percentError == 0);
+                assertTrue(memoryEstimator.isHighVolumeBuckets());
+            }
+
+            if (shouldReturn) {
+                o.setSizeInBytes(0);
+                memoryEstimator.reset();
+            }
+        }
+        assertTrue(shouldReturn);
+    }
+
+    private SortedMap<Key, Value> createTestData3(int elapsedTime, int skipInterval, int numTagVariations) {
+        SortedMap<Key, Value> testData3 = new TreeMap<>();
+        List<List<Tag>> listOfTagVariations = new ArrayList<>();
+        for (int x = 1; x <= numTagVariations; x++) {
+            listOfTagVariations.add(Collections.singletonList(new Tag("instance", Integer.toString(x))));
+        }
+
+        for (long i = 0; i < elapsedTime; i += skipInterval) {
+            for (List<Tag> tags : listOfTagVariations) {
+                put(testData3, new Metric("sys.loadAvg", i, i * 2, tags));
+            }
+        }
+        return testData3;
+    }
+
+    @Test
+    public void testDownsampleCombining() throws Exception {
+
+        int numTagVariations = 2;
+        int sampleInterval = 50;
+        int elapsedTime = 100;
+        int skipInterval = 10;
+        SortedMap<Key, Value> testData3 = createTestData3(elapsedTime, skipInterval, numTagVariations);
+        DownsampleIterator iter = new DownsampleIterator();
+        Map<Set<Tag>, Downsample> samples = runQuery(iter, testData3, sampleInterval, 1000);
+        assertEquals(numTagVariations, samples.size());
+        long totalBuckets = 0;
+        for (Entry<Set<Tag>, Downsample> entry : samples.entrySet()) {
+            totalBuckets = totalBuckets + entry.getValue().getNumBuckets();
+        }
+        assertEquals((elapsedTime / sampleInterval) * numTagVariations, totalBuckets);
+    }
+
     private Map<Set<Tag>, Downsample> runQuery(SortedKeyValueIterator<Key, Value> iter, SortedMap<Key, Value> testData,
-            long period) throws Exception {
+            long period, long maxDownsampleMemory) throws Exception {
         IteratorSetting is = new IteratorSetting(100, DownsampleIterator.class);
-        DownsampleIterator.setDownsampleOptions(is, 0, 1000, period, Avg.class.getName());
+        DownsampleIterator.setDownsampleOptions(is, 0, 1000, period, maxDownsampleMemory, Avg.class.getName());
         SortedKeyValueIterator<Key, Value> source = new SortedMapIterator(testData);
         iter.init(source, is.getOptions(), null);
         iter.seek(new Range(), Collections.emptyList(), true);
-        assertTrue(iter.hasTop());
-        Key key = iter.getTopKey();
+        boolean hasTop = iter.hasTop();
+        assertTrue(hasTop);
+        Key key = null;
+        Map<Set<Tag>, Downsample> samples = new HashMap<>();
+        do {
+            Map<Set<Tag>, Downsample> currentSamples = DownsampleIterator.decodeValue(iter.getTopValue());
+            List<Downsample> downsampleArray = new ArrayList<>();
+            for (Entry<Set<Tag>, Downsample> entry : currentSamples.entrySet()) {
+                Downsample downsample = samples.get(entry.getKey());
+                if (downsample == null) {
+                    samples.put(entry.getKey(), entry.getValue());
+                } else {
+                    downsampleArray.clear();
+                    downsampleArray.add(downsample);
+                    downsampleArray.add(entry.getValue());
+                    samples.put(entry.getKey(), Downsample.combineDownsample(downsampleArray, null));
+                }
+            }
+            key = iter.getTopKey();
+            System.out.println(key.toString());
+        } while (iter.hasTop());
+
         assertEquals(testData.lastKey(), key);
-        Map<Set<Tag>, Downsample> samples = DownsampleIterator.decodeValue(iter.getTopValue());
         return samples;
     }
 

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

@@ -22,6 +22,7 @@ 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 {
 
@@ -38,7 +39,8 @@ 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), new byte[0], new byte[0], ts);
+            Key k = new Key(row, tags.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(ts,
+                    ""), new byte[0], ts);
             Value v = new Value(MetricAdapter.encodeValue(m.getValue().getMeasure()));
             table.put(k, v);
         }
@@ -68,7 +70,8 @@ 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), new byte[0], new byte[0], ts);
+            Key k = new Key(row, tags.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(ts,
+                    ""), new byte[0], ts);
             Value v = new Value(MetricAdapter.encodeValue(m.getValue().getMeasure()));
             table.put(k, v);
         }
@@ -113,7 +116,8 @@ 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), new byte[0], new byte[0], ts);
+                Key k = new Key(row, tags.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(
+                        ts, ""), new byte[0], ts);
                 Value v = new Value(MetricAdapter.encodeValue(m.getValue().getMeasure()));
                 table.put(k, v);
             }
@@ -147,7 +151,8 @@ 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), new byte[0], new byte[0], ts);
+                Key k = new Key(row, tags.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(
+                        ts, ""), new byte[0], ts);
                 Value v = new Value(MetricAdapter.encodeValue(m.getValue().getMeasure()));
                 table.put(k, v);
             }
@@ -181,7 +186,8 @@ 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), new byte[0], new byte[0], ts);
+                Key k = new Key(row, tags.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(
+                        ts, ""), 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,10 +6,7 @@ import static org.junit.Assert.assertTrue;
 
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.TreeMap;
+import java.util.*;
 
 import org.apache.accumulo.core.client.IteratorSetting;
 import org.apache.accumulo.core.data.Key;
@@ -43,7 +40,8 @@ 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), new byte[0], new byte[0], ts);
+            Key k = new Key(row, tags.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(ts,
+                    ""), new byte[0], ts);
             Value v = new Value(MetricAdapter.encodeValue(m.getValue().getMeasure()));
             table.put(k, v);
         }
@@ -92,12 +90,14 @@ 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), new byte[0], new byte[0], ts);
+            Key k = new Key(row, tags1.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(ts,
+                    ""), 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), new byte[0], new byte[0], ts);
+            Key k2 = new Key(row2, tags2.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(
+                    ts, ""), new byte[0], ts);
             Value v2 = new Value(MetricAdapter.encodeValue(m2.getValue().getMeasure()));
             table.put(k2, v2);
         }
@@ -108,8 +108,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 });
+        }
+        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());
@@ -128,19 +132,21 @@ 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), new byte[0], new byte[0], ts);
+            Key k = new Key(row, tags1.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(ts,
+                    ""), 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), new byte[0], new byte[0],
-                        ts);
+                Key k2 = new Key(row2, tags2.get(0).join().getBytes(StandardCharsets.UTF_8),
+                        MetricAdapter.encodeColQual(ts, ""), 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);
@@ -148,12 +154,15 @@ 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 });
-            if (i < 50) {
-                checkNextResult(iter, new double[] { (i - 4) * 2, (i - 3) * 2, (i - 2) * 2, (i - 1) * 2, i * 2 });
-            }
+        }
+        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 });
         }
         assertFalse(iter.hasTop());
     }
@@ -170,15 +179,16 @@ 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), new byte[0], new byte[0], ts);
+            Key k = new Key(row, tags1.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(ts,
+                    ""), 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), new byte[0], new byte[0],
-                        ts);
+                Key k2 = new Key(row2, tags2.get(0).join().getBytes(StandardCharsets.UTF_8),
+                        MetricAdapter.encodeColQual(ts, ""), new byte[0], ts);
                 Value v2 = new Value(MetricAdapter.encodeValue(m2.getValue().getMeasure()));
                 table.put(k2, v2);
             }
@@ -190,12 +200,15 @@ 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());
 
     }
@@ -214,22 +227,25 @@ 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), new byte[0], new byte[0], ts);
+            Key k = new Key(row, tags1.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(ts,
+                    ""), 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), new byte[0], new byte[0],
-                    ts + 50);
+            Key k2 = new Key(row2, tags2.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(
+                    ts, ""), 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), new byte[0], new byte[0], ts);
+            Key k3 = new Key(row3, tags3.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(
+                    ts, ""), 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);
@@ -256,23 +272,20 @@ public class TimeSeriesGroupingIteratorTest extends IteratorTestBase {
         third.add(9D);
         third.add(12D);
 
-        for (int i = 4; i < 196; i++) {
+        // 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++) {
             System.out.println(i);
-            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);
-            }
+            checkNextResult(iter, second);
+            shiftAndAdd(second, 2);
+        }
+        for (int i = 4; i < 100; i++) {
+            checkNextResult(iter, third);
+            shiftAndAdd(third, 3);
         }
         assertFalse(iter.hasTop());
     }

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

@@ -49,6 +49,7 @@ 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,6 +572,11 @@ 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)
@@ -590,6 +595,11 @@ 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,6 +346,15 @@ 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());