SELECTED_LABELS option

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

@@ -11,6 +11,7 @@ public enum Keyword implements ProtocolCommand {
   RETENTION,
   TIMESTAMP,
   WITHLABELS,
+  SELECTED_LABELS,
   COUNT,
   UNCOMPRESSED,
   CHUNK_SIZE,

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

@@ -13,6 +13,7 @@ public class MultiRangeParams {
   private long timeBucket;
 
   private boolean withLabels;
+  private String[] selectedLabels;
 
   public static MultiRangeParams multiRangeParams() {
     return new MultiRangeParams();
@@ -41,6 +42,12 @@ public MultiRangeParams withLabels(boolean withLabels) {
     return this;
   }
 
+  /** WARNING: SELECTED_LABELS and WITHLABELS are mutually exclusive. */
+  public MultiRangeParams selectedLabels(String... labels) {
+    this.selectedLabels = labels;
+    return this;
+  }
+
   public byte[][] getByteParams(Long from, Long to, String... filters) {
     List<byte[]> params = new ArrayList<>();
 
@@ -67,7 +74,12 @@ public byte[][] getByteParams(Long from, Long to, String... filters) {
       params.add(Protocol.toByteArray(timeBucket));
     }
 
-    if (withLabels) {
+    if (selectedLabels != null) {
+      params.add(Keyword.SELECTED_LABELS.getRaw());
+      for (String label : selectedLabels) {
+        params.add(SafeEncoder.encode(label));
+      }
+    } else if (withLabels) {
       params.add(Keyword.WITHLABELS.getRaw());
     }
 

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

@@ -43,8 +43,10 @@ protected static Value[] parseRange(List<Object> range) {
   private static Map<String, String> getLabelsStringStringMap(List<?> resLabels) {
     Map<String, String> rangeLabels = new HashMap<>(resLabels.size());
     for (Object resLabel : resLabels) {
-      List<byte[]> label = (List<byte[]>) resLabel;
-      rangeLabels.put(SafeEncoder.encode(label.get(0)), SafeEncoder.encode(label.get(1)));
+      List<byte[]> pair = (List<byte[]>) resLabel;
+      if (pair.get(0) != null && pair.get(1) != null) {
+        rangeLabels.put(SafeEncoder.encode(pair.get(0)), SafeEncoder.encode(pair.get(1)));
+      }
     }
     return rangeLabels;
   }

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

@@ -416,6 +416,56 @@ public void testIncrByDecrBy() throws InterruptedException {
     }
   }
 
+  @Test
+  public void mrangeSelectedLabels() {
+
+    client.create("ts1", convertMap("l1", "v1", "l2", "v2"));
+    client.create("ts2", convertMap("l1", "v1", "l3", "v3"));
+
+    Value[] rawValues =
+        new Value[] {
+          new Value(1000L, 1.0),
+          new Value(2000L, 0.9),
+          new Value(3200L, 1.1),
+          new Value(4500L, -1.1)
+        };
+
+    client.add("ts1", rawValues[0].getTime(), rawValues[0].getValue());
+    client.add("ts2", rawValues[1].getTime(), rawValues[1].getValue());
+    client.add("ts2", rawValues[2].getTime(), rawValues[2].getValue());
+    client.add("ts1", rawValues[3].getTime(), rawValues[3].getValue());
+
+    Range[] range =
+        client.mrange(0L, 5000L, MultiRangeParams.multiRangeParams().withLabels(), "l1=v1");
+    assertEquals(2, range.length);
+    assertEquals("ts1", range[0].getKey());
+    assertEquals(2, range[0].getLabels().size());
+    assertEquals("ts2", range[1].getKey());
+    assertEquals(2, range[1].getLabels().size());
+
+    range =
+        client.mrange(
+            0L, 5000L, MultiRangeParams.multiRangeParams().selectedLabels("l1", "l2"), "l1=v1");
+    assertEquals(2, range.length);
+    assertEquals("ts1", range[0].getKey());
+    assertEquals(2, range[0].getLabels().size());
+    assertEquals("ts2", range[1].getKey());
+    assertEquals(1, range[1].getLabels().size());
+
+    range =
+        client.mrange(0L, 5000L, MultiRangeParams.multiRangeParams().selectedLabels("l3"), "l2=v2");
+    assertEquals(1, range.length);
+    assertEquals(0, range[0].getLabels().size());
+  }
+
+  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() {