Feature: Sliding Windows (#515)

* Store timestamp in state value

* Decouple state operations from aggregate functions

* Check max expired window start instead of min valid window end

* Pass watermark to expire_windows

* Decouple window deletion from window expiration

* Feature: Sliding Windows

* Do not emit right windows via .final()

* Do not emit right windows via .current()

* Move SlidingWindow to a separate module

* Don't use deque

* Replace watermark with max_start_time

* Set expiration watermark only once

* Remove timeit tests

These were complicated. Custom performance tests are better.

* Refactoring tests and fixing problems

* Test presence of exact windows in the state

* Log late_by_ms from FixedTimeWindow

* Rename watermark variables

* Log expired windows from SlidingWindow

* Correct existing windowing docs

Corrected timestamp keys and changed temperature readings to numbers
that are easier to calculate without calculator and also look more
familiar in both Celsius and Fahrenheit scales.

* Add sliding windows docs

* Create docstring describing sliding window algorithm

* Fix imports

* Fix TestWindowedRocksDBPartitionTransaction after rebase

* Latest deleted window timestamps per key

* Create helper TimestampsCache class

Author: gwaramadze

docs/windowing.md

@@ -167,17 +167,17 @@ Input:
 
 (Here the `"timestamp"` column illustrates Kafka message timestamps)
 ```json
-{"temperature": 65, "timestamp": 100}
-{"temperature": 52, "timestamp": 200}
-{"temperature": 61, "timestamp": 300}
+{"temperature": 30, "timestamp": 100}
+{"temperature": 29, "timestamp": 200}
+{"temperature": 28, "timestamp": 300}
 ```
 
 Expected output:
 
 ```json
-{"avg_temperature": 65, "window_start": 0, "window_end": 3600000}
-{"avg_temperature": 58.5, "window_start": 0, "window_end": 3600000}
-{"avg_temperature": 59.333, "window_start": 0, "window_end": 3600000}
+{"avg_temperature": 30, "window_start_ms": 0, "window_end_ms": 3600000}
+{"avg_temperature": 29.5, "window_start_ms": 0, "window_end_ms": 3600000}
+{"avg_temperature": 29, "window_start_ms": 0, "window_end_ms": 3600000}
 ```
 
 Here is how to do it using tumbling windows: 
@@ -253,21 +253,21 @@ Input:
 (Here the `"timestamp"` column illustrates Kafka message timestamps)
 
 ```json
-{"temperature": 65, "timestamp": 50000}
-{"temperature": 52, "timestamp": 60000}
-{"temperature": 61, "timestamp": 62000}
+{"temperature": 30, "timestamp": 50000}
+{"temperature": 29, "timestamp": 60000}
+{"temperature": 28, "timestamp": 62000}
 ```
 
 Expected output:
 
 ```json
-{"avg_temperature": 65, "window_start": 0, "window_end": 3600000}
+{"avg_temperature": 30, "window_start_ms": 0, "window_end_ms": 3600000}
 
-{"avg_temperature": 58.5, "window_start": 0, "window_end": 3600000}
-{"avg_temperature": 65, "window_start": 60000, "window_end": 4200000}
+{"avg_temperature": 29.5, "window_start_ms": 0, "window_end_ms": 3600000}
+{"avg_temperature": 30, "window_start_ms": 60000, "window_end_ms": 4200000}
 
-{"avg_temperature": 59.333, "window_start": 0, "window_end": 3600000}
-{"avg_temperature": 56.5, "window_start": 60000, "window_end": 4200000}
+{"avg_temperature": 29, "window_start_ms": 0, "window_end_ms": 3600000}
+{"avg_temperature": 28.5, "window_start_ms": 60000, "window_end_ms": 4200000}
 ```
 
 
@@ -305,6 +305,94 @@ sdf = (
 ```
 
 
+## Sliding Windows
+Sliding windows are overlapping time-based windows that advance with each incoming message, rather than at fixed time intervals like hopping windows. They have a fixed 1 ms resolution and perform better and are less resource-intensive than hopping windows with a 1 ms step. Sliding windows do not produce redundant windows; every interval has a distinct aggregation.
+
+Sliding windows provide optimal performance for tasks requiring high-precision real-time monitoring. However, if the task is not time-critical or the data stream is extremely dense, tumbling or hopping windows may perform better.
+
+For example, a sliding window of 1 hour will generate the following intervals as messages A, B, C, and D arrive:
+
+```
+                Sliding Windows
+
+Time
+[00:00:00.000, 01:00:00.000):  ......A
+[00:00:00.001, 01:00:00.001):   ......B
+[00:00:00.003, 01:00:00.003):     ......C
+[00:00:00.006, 01:00:00.006):        ......D
+
+```
+   
+Note that both the start and the end of the interval are inclusive.
+
+In sliding windows, each timestamp can be assigned to multiple intervals because these intervals overlap.
+
+For example, a timestamp `01:33:13.000` will match intervals for all messages incoming between `01:33:13.000` and `02:33:13.000`. Borderline windows including this timestamp will be:
+
+   - `00:33:13.000 - 01:33:13.000`
+   - `01:33:13.000 - 02:33:13.000`
+
+    
+**Example:**
+
+Imagine you receive temperature readings from sensors, and you need to calculate the average temperature for the last hour, producing updates for each incoming message. The message key is a sensor ID, so the aggregations will be grouped by each sensor.
+
+Input:  
+(Here the `"timestamp"` column illustrates Kafka message timestamps)
+
+```json
+{"temperature": 30, "timestamp": 3600000}
+{"temperature": 29, "timestamp": 4800000}
+{"temperature": 28, "timestamp": 4800001}
+{"temperature": 27, "timestamp": 7200000}
+{"temperature": 26, "timestamp": 7200001}
+```
+
+Expected output:
+
+```json
+{"avg_temperature": 30, "window_start_ms": 0, "window_end_ms": 3600000}
+{"avg_temperature": 29.5, "window_start_ms": 1200000, "window_end_ms": 4800000}
+{"avg_temperature": 29, "window_start_ms": 1200001, "window_end_ms": 4800001}
+{"avg_temperature": 28.5, "window_start_ms": 3600000, "window_end_ms": 7200000}
+{"avg_temperature": 27.5, "window_start_ms": 3600001, "window_end_ms": 7200001}  # reading 30 is outside of the window
+```
+
+
+```python
+from datetime import timedelta
+from quixstreams import Application
+
+app = Application(...)
+sdf = app.dataframe(...)
+
+sdf = (
+    # Extract "temperature" value from the message
+    sdf.apply(lambda value: value["temperature"])
+
+    # Define a sliding window of 1h
+    # You can also pass duration_ms as integer of milliseconds
+    .sliding_window(duration_ms=timedelta(hours=1))
+
+    # Specify the "mean" aggregate function
+    .mean()
+
+    # Emit updates for each incoming message
+    .current()
+
+    # Unwrap the aggregated result to match the expected output format
+    .apply(
+        lambda result: {
+            "avg_temperature": result["value"],
+            "window_start_ms": result["start"],
+            "window_end_ms": result["end"],
+        }
+    )
+)
+
+```
+
+
 ## Supported Aggregations
 
 Currently, windows support the following aggregation functions:

quixstreams/dataframe/dataframe.py

@@ -52,7 +52,11 @@
 from .registry import DataframeRegistry
 from .series import StreamingSeries
 from .utils import ensure_milliseconds
-from .windows import HoppingWindowDefinition, TumblingWindowDefinition
+from .windows import (
+    HoppingWindowDefinition,
+    SlidingWindowDefinition,
+    TumblingWindowDefinition,
+)
 
 ApplyCallbackStateful = Callable[[Any, State], Any]
 ApplyWithMetadataCallbackStateful = Callable[[Any, Any, int, Any, State], Any]
@@ -821,11 +825,11 @@ def tumbling_window(
             .sum()
 
             # Specify how the results should be emitted downstream.
-            # "all()" will emit results as they come for each updated window,
+            # "current()" will emit results as they come for each updated window,
             # possibly producing multiple messages per key-window pair
             # "final()" will emit windows only when they are closed and cannot
             # receive any updates anymore.
-            .all()
+            .current()
         )
         ```
 
@@ -900,11 +904,11 @@ def hopping_window(
             .sum()
 
             # Specify how the results should be emitted downstream.
-            # "all()" will emit results as they come for each updated window,
+            # "current()" will emit results as they come for each updated window,
             # possibly producing multiple messages per key-window pair
             # "final()" will emit windows only when they are closed and cannot
             # receive any updates anymore.
-            .all()
+            .current()
         )
         ```
 
@@ -951,6 +955,87 @@ def hopping_window(
             name=name,
         )
 
+    def sliding_window(
+        self,
+        duration_ms: Union[int, timedelta],
+        grace_ms: Union[int, timedelta] = 0,
+        name: Optional[str] = None,
+    ) -> SlidingWindowDefinition:
+        """
+        Create a sliding window transformation on this StreamingDataFrame.
+        Sliding windows continuously evaluate the stream with a fixed step of 1 ms
+        allowing for overlapping, but not redundant windows of a fixed size.
+
+        Sliding windows are similar to hopping windows with step_ms set to 1,
+        but are siginificantly more perforant.
+
+        They allow performing stateful aggregations like `sum`, `reduce`, etc.
+        on top of the data and emit results downstream.
+
+        Notes:
+
+        - The timestamp of the aggregation result is set to the window start timestamp.
+        - Every window is grouped by the current Kafka message key.
+        - Messages with `None` key will be ignored.
+        - The time windows always use the current event time.
+        - Windows are inclusive on both the start end end time.
+        - Every window contains a distinct aggregation.
+
+        Example Snippet:
+
+        ```python
+        app = Application()
+        sdf = app.dataframe(...)
+
+        sdf = (
+            # Define a sliding window of 60s with a grace period of 10s
+            sdf.sliding_window(
+                duration_ms=timedelta(seconds=60),
+                grace_ms=timedelta(seconds=10)
+            )
+
+            # Specify the aggregation function
+            .sum()
+
+            # Specify how the results should be emitted downstream.
+            # "current()" will emit results as they come for each updated window,
+            # possibly producing multiple messages per key-window pair
+            # "final()" will emit windows only when they are closed and cannot
+            # receive any updates anymore.
+            .current()
+        )
+        ```
+
+        :param duration_ms: The length of each window.
+            Can be specified as either an `int` representing milliseconds or a
+            `timedelta` object.
+            >***NOTE:*** `timedelta` objects will be rounded to the closest millisecond
+            value.
+
+        :param grace_ms: The grace period for data arrival.
+            It allows late-arriving data (data arriving after the window
+            has theoretically closed) to be included in the window.
+            Can be specified as either an `int` representing milliseconds
+            or as a `timedelta` object.
+            >***NOTE:*** `timedelta` objects will be rounded to the closest millisecond
+            value.
+
+        :param name: The unique identifier for the window. If not provided, it will be
+            automatically generated based on the window's properties.
+
+        :return: `SlidingWindowDefinition` instance representing the sliding window
+            configuration.
+            This object can be further configured with aggregation functions
+            like `sum`, `count`, etc. applied to the StreamingDataFrame.
+        """
+
+        duration_ms = ensure_milliseconds(duration_ms)
+        grace_ms = ensure_milliseconds(grace_ms)
+
+        return SlidingWindowDefinition(
+            duration_ms=duration_ms, grace_ms=grace_ms, dataframe=self, name=name
+        )
+
     def drop(
         self,
         columns: Union[str, List[str]],

quixstreams/dataframe/windows/__init__.py

@@ -1,8 +1,13 @@
 from .base import WindowResult
-from .definitions import HoppingWindowDefinition, TumblingWindowDefinition
+from .definitions import (
+    HoppingWindowDefinition,
+    SlidingWindowDefinition,
+    TumblingWindowDefinition,
+)
 
 __all__ = [
     "HoppingWindowDefinition",
+    "SlidingWindowDefinition",
     "TumblingWindowDefinition",
     "WindowResult",
 ]

quixstreams/dataframe/windows/base.py

@@ -3,16 +3,14 @@
 
 from typing_extensions import TypedDict
 
-from quixstreams.state import WindowedState
-
 
 class WindowResult(TypedDict):
     start: int
     end: int
     value: Any
 
 
-WindowAggregateFunc = Callable[[int, int, int, Any, WindowedState], Any]
+WindowAggregateFunc = Callable[[Any, Any], Any]
 WindowMergeFunc = Callable[[Any], Any]