Add time window and window assigner

core/src/main/java/org/opensearch/sql/planner/streaming/windowing/Window.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.planner.streaming.windowing;
+
+import lombok.Data;
+
+/**
+ * A time window is a window of time interval with inclusive start time and exclusive end time.
+ */
+@Data
+public class Window {
+
+  /** Start timestamp (inclusive) of the time window. */
+  private final long startTime;
+
+  /** End timestamp (exclusive) of the time window. */
+  private final long endTime;
+
+  /**
+   * Return the maximum timestamp (inclusive) of the window.
+   */
+  public long maxTimestamp() {
+    return endTime - 1;
+  }
+}

core/src/main/java/org/opensearch/sql/planner/streaming/windowing/assigner/SlidingWindowAssigner.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.planner.streaming.windowing.assigner;
+
+import com.google.common.base.Preconditions;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import org.opensearch.sql.planner.streaming.windowing.Window;
+
+/**
+ * A sliding window assigner assigns multiple overlapped window per event timestamp.
+ * The overlap size is determined by the given slide interval.
+ */
+public class SlidingWindowAssigner implements WindowAssigner {
+
+  /** Window size in millisecond. */
+  private final long windowSize;
+
+  /** Slide size in millisecond. */
+  private final long slideSize;
+
+  /**
+   * Create sliding window assigner with the given window and slide size.
+   */
+  public SlidingWindowAssigner(long windowSize, long slideSize) {
+    Preconditions.checkArgument(windowSize > 0,
+        "Window size [%s] must be positive number", windowSize);
+    Preconditions.checkArgument(slideSize > 0,
+        "Slide size [%s] must be positive number", slideSize);
+    this.windowSize = windowSize;
+    this.slideSize = slideSize;
+  }
+
+  @Override
+  public List<Window> assign(long timestamp) {
+    List<Window> windows = new ArrayList<>();
+
+    // Assign window from the last start time to first until given timestamp outside current window
+    long startTime = timestamp - timestamp % slideSize;
+    for (Window win = window(startTime); win.maxTimestamp() >= timestamp; win = window(startTime)) {
+      windows.add(win);
+      startTime -= slideSize;
+    }
+
+    // Reverse the window list for easy read and test
+    Collections.reverse(windows);
+    return windows;
+  }
+
+  private Window window(long startTime) {
+    return new Window(startTime, startTime + windowSize);
+  }
+}

core/src/main/java/org/opensearch/sql/planner/streaming/windowing/assigner/TumblingWindowAssigner.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.planner.streaming.windowing.assigner;
+
+import com.google.common.base.Preconditions;
+import java.util.Collections;
+import java.util.List;
+import org.opensearch.sql.planner.streaming.windowing.Window;
+
+/**
+ * A tumbling window assigner assigns a single window per event timestamp
+ * without overlap.
+ */
+public class TumblingWindowAssigner implements WindowAssigner {
+
+  /** Window size in millisecond. */
+  private final long windowSize;
+
+  /**
+   * Create tumbling window assigner with the given window size.
+   */
+  public TumblingWindowAssigner(long windowSize) {
+    Preconditions.checkArgument(windowSize > 0,
+        "Window size [%s] must be positive number", windowSize);
+    this.windowSize = windowSize;
+  }
+
+  @Override
+  public List<Window> assign(long timestamp) {
+    long startTime = timestamp - timestamp % windowSize;
+    return Collections.singletonList(new Window(startTime, startTime + windowSize));
+  }
+}

core/src/main/java/org/opensearch/sql/planner/streaming/windowing/assigner/WindowAssigner.java

@@ -0,0 +1,24 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.planner.streaming.windowing.assigner;
+
+import java.util.List;
+import org.opensearch.sql.planner.streaming.windowing.Window;
+
+/**
+ * A window assigner assigns zero or more window to an event timestamp
+ * based on different windowing approach.
+ */
+public interface WindowAssigner {
+
+  /**
+   * Return window(s) assigned to the timestamp.
+   * @param timestamp given event timestamp
+   * @return windows assigned
+   */
+  List<Window> assign(long timestamp);
+
+}

core/src/test/java/org/opensearch/sql/planner/streaming/windowing/WindowTest.java

@@ -0,0 +1,21 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.planner.streaming.windowing;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import org.junit.jupiter.api.Test;
+
+class WindowTest {
+
+  @Test
+  void test() {
+    Window window = new Window(1000, 2000);
+    assertEquals(1000, window.getStartTime());
+    assertEquals(2000, window.getEndTime());
+    assertEquals(1999, window.maxTimestamp());
+  }
+}

core/src/test/java/org/opensearch/sql/planner/streaming/windowing/assigner/SlidingWindowAssignerTest.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.planner.streaming.windowing.assigner;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.util.List;
+import org.junit.jupiter.api.Test;
+import org.opensearch.sql.planner.streaming.windowing.Window;
+
+class SlidingWindowAssignerTest {
+
+  @Test
+  void testAssignWindows() {
+    long windowSize = 1000;
+    long slideSize = 500;
+    SlidingWindowAssigner assigner = new SlidingWindowAssigner(windowSize, slideSize);
+
+    assertEquals(
+        List.of(
+            new Window(0, 1000),
+            new Window(500, 1500)),
+        assigner.assign(500));
+
+    assertEquals(
+        List.of(
+            new Window(0, 1000),
+            new Window(500, 1500)),
+        assigner.assign(999));
+
+    assertEquals(
+        List.of(
+            new Window(500, 1500),
+            new Window(1000, 2000)),
+        assigner.assign(1000));
+  }
+
+  @Test
+  void testConstructWithIllegalArguments() {
+    IllegalArgumentException error1 = assertThrows(IllegalArgumentException.class,
+        () -> new SlidingWindowAssigner(-1, 100));
+    assertEquals("Window size [-1] must be positive number", error1.getMessage());
+
+    IllegalArgumentException error2 = assertThrows(IllegalArgumentException.class,
+        () -> new SlidingWindowAssigner(1000, 0));
+    assertEquals("Slide size [0] must be positive number", error2.getMessage());
+  }
+}

core/src/test/java/org/opensearch/sql/planner/streaming/windowing/assigner/TumblingWindowAssignerTest.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.planner.streaming.windowing.assigner;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.util.Collections;
+import org.junit.jupiter.api.Test;
+import org.opensearch.sql.planner.streaming.windowing.Window;
+
+class TumblingWindowAssignerTest {
+
+  @Test
+  void testAssignWindow() {
+    long windowSize = 1000;
+    TumblingWindowAssigner assigner = new TumblingWindowAssigner(windowSize);
+
+    assertEquals(
+        Collections.singletonList(new Window(0, 1000)),
+        assigner.assign(500));
+    assertEquals(
+        Collections.singletonList(new Window(1000, 2000)),
+        assigner.assign(1999));
+    assertEquals(
+        Collections.singletonList(new Window(2000, 3000)),
+        assigner.assign(2000));
+  }
+
+  @Test
+  void testConstructWithIllegalWindowSize() {
+    IllegalArgumentException error = assertThrows(IllegalArgumentException.class,
+        () -> new TumblingWindowAssigner(-1));
+    assertEquals("Window size [-1] must be positive number", error.getMessage());
+  }
+}