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

client/pom.xml

@@ -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;
+            }
+        }
+    }
+
+}

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/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/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.apache.log4j.Logger;
 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;
@@ -30,20 +29,125 @@ public class DownsampleIterator extends WrappingIterator {
     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 MemoryEstimator memoryEstimator = null;
+
+    static public class MemoryEstimator {
+
+        private boolean newBucket = false;
+        private long start;
+        private long period;
+        private long startOfCurrentBucket;
+        private long bucketsCompleted = 0;
+        private long bytesPerBucket = 0;
+        boolean highVolumeBuckets = false;
+        long maxDownsampleMemory = 0; // max aggregation memory (bytes) before
+                                      // current batch is returned (after
+                                      // bucket is complete)
+        LinkedList<Long> percentageChecks = new LinkedList<>();
+
+        public void reset() {
+            newBucket = false;
+            bucketsCompleted = 0;
+            bytesPerBucket = 0;
+            percentageChecks.add(5l);
+            percentageChecks.add(25l);
+            percentageChecks.add(50l);
+            percentageChecks.add(75l);
+        }
+
+        public MemoryEstimator(long maxDownsampleMemory, long start, long period) {
+            this.maxDownsampleMemory = maxDownsampleMemory;
+            this.start = start;
+            this.period = period;
+            this.startOfCurrentBucket = this.start;
+            reset();
+        }
+
+        public double getMemoryUsedPercentage() {
+            return (bucketsCompleted * bytesPerBucket) / (double) maxDownsampleMemory * 100;
+        }
+
+        public long getBucketsCompleted() {
+            return bucketsCompleted;
+        }
+
+        public long getBytesPerBucket() {
+            return bytesPerBucket;
+        }
+
+        public boolean isHighVolumeBuckets() {
+            return highVolumeBuckets;
+        }
+
+        public boolean shouldReturnBasedOnMemoryUsage(long timestamp, Object value) {
+
+            boolean shouldReturn = false;
+            if (maxDownsampleMemory >= 0) {
+                sample(timestamp);
+                if (isNewBucket()) {
+                    bucketsCompleted++;
+                    double memoryUsedPercentage = getMemoryUsedPercentage();
+                    if (memoryUsedPercentage >= 100) {
+                        shouldReturn = true;
+                    } else {
+                        boolean checkMemoryNow = false;
+                        Long check = percentageChecks.peek();
+                        if (check != null && memoryUsedPercentage >= check) {
+                            checkMemoryNow = true;
+                            percentageChecks.removeFirst();
+                        }
+                        // recalculate bytesPerBucket
+                        if (bytesPerBucket == 0 || highVolumeBuckets || checkMemoryNow) {
+                            long memoryUsed = ObjectSizeOf.Sizer.getObjectSize(value);
+                            bytesPerBucket = memoryUsed / bucketsCompleted;
+                        }
+                        // bucket average greater than 10% of max
+                        highVolumeBuckets = (bytesPerBucket / (double) maxDownsampleMemory) >= 0.1;
+                        shouldReturn = false;
+                    }
+                }
+            }
+            return shouldReturn;
+        }
+
+        public boolean isNewBucket() {
+            return newBucket;
+        }
+
+        private void sample(long timestamp) {
+            if (timestamp >= (startOfCurrentBucket + period)) {
+                newBucket = true;
+                startOfCurrentBucket = timestamp - ((timestamp - start) % period);
+            } else {
+                newBucket = false;
+            }
+        }
+    }
 
     @SuppressWarnings("unchecked")
     @Override
     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 MemoryEstimator(maxDownsampleMemory, start, period);
+
         String aggClassname = options.get(AGGCLASS);
         Class<?> aggClass;
         try {
@@ -56,26 +160,38 @@ 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()) {
+            while (super.hasTop()) {
+                Key topKey = super.getTopKey();
+                Value topValue = super.getTopValue();
+                Metric metric = MetricAdapter.parse(topKey, topValue);
+                long timestamp = metric.getValue().getTimestamp();
+                if (memoryEstimator.shouldReturnBasedOnMemoryUsage(timestamp, value)) {
+                    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());
                 try {
-                    value.put(tags, sample = factory.create());
-                } catch (TimelyException e) {
-                    throw new RuntimeException(e);
+                    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);
-            }
+            return last != null;
+        } else {
+            return false;
         }
-        return last != null;
     }
 
     @Override
@@ -89,6 +205,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 +216,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/store/DataStoreImpl.java

@@ -125,6 +125,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 +150,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();
@@ -506,7 +508,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);
                     }
 

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;
@@ -72,6 +69,7 @@ public class DownsampleIteratorTest {
     }
 
     private void createTestData2() {
+        List<List<Tag>> listOfTagVariations = new ArrayList<>();
         List<Tag> tags = Collections.singletonList(new Tag("host", "host1"));
         List<Tag> tags2 = Collections.singletonList(new Tag("host", "host2"));
 
@@ -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,143 @@ 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();
+        DownsampleIterator.MemoryEstimator memoryEstimator = new DownsampleIterator.MemoryEstimator(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();
+        DownsampleIterator.MemoryEstimator memoryEstimator = new DownsampleIterator.MemoryEstimator(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();
+        } while (iter.hasTop());
+
         assertEquals(testData.lastKey(), key);
-        Map<Set<Tag>, Downsample> samples = DownsampleIterator.decodeValue(iter.getTopValue());
         return samples;
     }