ALIGN option (#69)

* ALIGN option * fix due to master * use constant Co-authored-by: Guy Korland <gkorland@gmail.com>
RedisTimeSeries · Aug 3, 2021 · 0ef81fc · 0ef81fc
1 parent 3c30148
commit 0ef81fc
Show file tree

Hide file tree

Showing 3 changed files with 64 additions and 2 deletions.
diff --git a/src/main/java/com/redislabs/redistimeseries/Keyword.java b/src/main/java/com/redislabs/redistimeseries/Keyword.java
@@ -17,6 +17,7 @@ public enum Keyword implements ProtocolCommand {
   CHUNK_SIZE,
   DUPLICATE_POLICY,
   ON_DUPLICATE,
+  ALIGN,
   FILTER_BY_TS,
   FILTER_BY_VALUE,
   GROUPBY,

diff --git a/src/main/java/com/redislabs/redistimeseries/RangeParams.java b/src/main/java/com/redislabs/redistimeseries/RangeParams.java
@@ -12,6 +12,8 @@ public class RangeParams {
 
   private Integer count;
 
+  private byte[] align;
+
   private Aggregation aggregationType;
   private long timeBucket;
 
@@ -34,6 +36,23 @@ public RangeParams count(int count) {
     return this;
   }
 
+  private RangeParams align(byte[] raw) {
+    this.align = raw;
+    return this;
+  }
+
+  public RangeParams align(long timestamp) {
+    return align(Protocol.toByteArray(timestamp));
+  }
+
+  public RangeParams alignStart() {
+    return align(RedisTimeSeries.MINUS);
+  }
+
+  public RangeParams alignEnd() {
+    return align(RedisTimeSeries.PLUS);
+  }
+
   public RangeParams aggregation(Aggregation aggregation, long timeBucket) {
     this.aggregationType = aggregation;
     this.timeBucket = timeBucket;
@@ -45,13 +64,13 @@ public byte[][] getByteParams(String key, Long from, Long to) {
     params.add(SafeEncoder.encode(key));
 
     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));
     }
@@ -75,6 +94,11 @@ public byte[][] getByteParams(String key, Long from, Long to) {
       params.add(Protocol.toByteArray(count));
     }
 
+    if (align != null) {
+      params.add(Keyword.ALIGN.getRaw());
+      params.add(align);
+    }
+
     if (aggregationType != null) {
       params.add(Keyword.AGGREGATION.getRaw());
       params.add(aggregationType.getRaw());

diff --git a/src/test/java/com/redislabs/redistimeseries/RedisTimeSeriesTest.java b/src/test/java/com/redislabs/redistimeseries/RedisTimeSeriesTest.java
@@ -416,6 +416,43 @@ public void testIncrByDecrBy() throws InterruptedException {
     }
   }
 
+  @Test
+  public void align() {
+    client.add("align", 1, 10d);
+    client.add("align", 3, 5d);
+    client.add("align", 11, 10d);
+    client.add("align", 25, 11d);
+
+    Value[] values =
+        client.range(
+            "align", 1L, 30L, RangeParams.rangeParams().aggregation(Aggregation.COUNT, 10));
+    assertArrayEquals(new Value[] {new Value(1, 2), new Value(11, 1), new Value(21, 1)}, values);
+
+    values =
+        client.range(
+            "align",
+            1L,
+            30L,
+            RangeParams.rangeParams().alignStart().aggregation(Aggregation.COUNT, 10));
+    assertArrayEquals(new Value[] {new Value(1, 2), new Value(11, 1), new Value(21, 1)}, values);
+
+    values =
+        client.range(
+            "align",
+            1L,
+            30L,
+            RangeParams.rangeParams().alignEnd().aggregation(Aggregation.COUNT, 10));
+    assertArrayEquals(new Value[] {new Value(1, 2), new Value(11, 1), new Value(21, 1)}, values);
+
+    values =
+        client.range(
+            "align",
+            1L,
+            30L,
+            RangeParams.rangeParams().align(5).aggregation(Aggregation.COUNT, 10));
+    assertArrayEquals(new Value[] {new Value(1, 2), new Value(11, 1), new Value(21, 1)}, values);
+  }
+
   @Test
   public void rangeFilterBy() {