Support GROUPBY and REDUCE options

src/main/java/com/redislabs/redistimeseries/Keyword.java

@@ -15,7 +15,9 @@ public enum Keyword implements ProtocolCommand {
   UNCOMPRESSED,
   CHUNK_SIZE,
   DUPLICATE_POLICY,
-  ON_DUPLICATE;
+  ON_DUPLICATE,
+  GROUPBY,
+  REDUCE;
 
   private final byte[] raw;
 

src/main/java/com/redislabs/redistimeseries/MultiRangeParams.java

@@ -14,6 +14,8 @@ public class MultiRangeParams {
 
   private boolean withLabels;
 
+  private String[] groupByReduce;
+
   public static MultiRangeParams multiRangeParams() {
     return new MultiRangeParams();
   }
@@ -41,17 +43,22 @@ public MultiRangeParams withLabels(boolean withLabels) {
     return this;
   }
 
+  public MultiRangeParams groupByReduce(String group, String reduce) {
+    this.groupByReduce = new String[] {group, reduce};
+    return this;
+  }
+
   public byte[][] getByteParams(Long from, Long to, String... filters) {
     List<byte[]> params = new ArrayList<>();
 
     if (from == null) {
-      params.add("-".getBytes());
+      params.add(RedisTimeSeries.MINUS);
     } else {
       params.add(Protocol.toByteArray(from));
     }
 
     if (to == null) {
-      params.add("+".getBytes());
+      params.add(RedisTimeSeries.PLUS);
     } else {
       params.add(Protocol.toByteArray(to));
     }
@@ -76,6 +83,13 @@ public byte[][] getByteParams(Long from, Long to, String... filters) {
       params.add(SafeEncoder.encode(filter));
     }
 
+    if (groupByReduce != null) {
+      params.add(Keyword.GROUPBY.getRaw());
+      params.add(SafeEncoder.encode(groupByReduce[0]));
+      params.add(Keyword.REDUCE.getRaw());
+      params.add(SafeEncoder.encode(groupByReduce[1]));
+    }
+
     return params.toArray(new byte[params.size()][]);
   }
 }

src/main/java/com/redislabs/redistimeseries/RedisTimeSeries.java

@@ -17,6 +17,8 @@
 public class RedisTimeSeries implements AutoCloseable {
 
   private static final byte[] STAR = SafeEncoder.encode("*");
+  protected static final byte[] PLUS = SafeEncoder.encode("+");
+  protected static final byte[] MINUS = SafeEncoder.encode("-");
 
   private final Pool<Jedis> pool;
 

src/test/java/com/redislabs/redistimeseries/RedisTimeSeriesTest.java

@@ -416,6 +416,44 @@ public void testIncrByDecrBy() throws InterruptedException {
     }
   }
 
+  @Test
+  public void groupByReduce() {
+    client.create("ts1", convertMap("metric", "cpu", "metric_name", "system"));
+    client.create("ts2", convertMap("metric", "cpu", "metric_name", "user"));
+
+    client.add("ts1", 1L, 90.0);
+    client.add("ts1", 2L, 45.0);
+    client.add("ts2", 2L, 99.0);
+
+    Range[] range =
+        client.mrange(
+            0L,
+            100L,
+            MultiRangeParams.multiRangeParams().withLabels().groupByReduce("metric_name", "max"),
+            "metric=cpu");
+    assertEquals(2, range.length);
+
+    assertEquals("metric_name=system", range[0].getKey());
+    assertEquals("system", range[0].getLabels().get("metric_name"));
+    assertEquals("max", range[0].getLabels().get("__reducer__"));
+    assertEquals("ts1", range[0].getLabels().get("__source__"));
+    assertArrayEquals(new Value[] {new Value(1, 90), new Value(2, 45)}, range[0].getValues());
+
+    assertEquals("metric_name=user", range[1].getKey());
+    assertEquals("user", range[1].getLabels().get("metric_name"));
+    assertEquals("max", range[1].getLabels().get("__reducer__"));
+    assertEquals("ts2", range[1].getLabels().get("__source__"));
+    assertArrayEquals(new Value[] {new Value(2, 99)}, range[1].getValues());
+  }
+
+  private Map<String, String> convertMap(String... array) {
+    Map<String, String> map = new HashMap<>(array.length / 2);
+    for (int i = 0; i < array.length; i += 2) {
+      map.put(array[i], array[i + 1]);
+    }
+    return map;
+  }
+
   @Test
   public void testGet() {