Merge pull request #360 from chinmaychandak/windowing_fix

Fix windowing over time bugs

Author: martindurant

streamz/dataframe/aggregations.py

@@ -167,18 +167,20 @@ def diff_iloc(dfs, new, window=None):
         List of dataframes to decay
     """
     dfs = deque(dfs)
-    dfs.append(new)
+    if len(new) > 0:
+        dfs.append(new)
     old = []
-    n = sum(map(len, dfs)) - window
-    while n > 0:
-        if len(dfs[0]) <= n:
-            df = dfs.popleft()
-            old.append(df)
-            n -= len(df)
-        else:
-            old.append(dfs[0].iloc[:n])
-            dfs[0] = dfs[0].iloc[n:]
-            n = 0
+    if len(dfs) > 0:
+        n = sum(map(len, dfs)) - window
+        while n > 0:
+            if len(dfs[0]) <= n:
+                df = dfs.popleft()
+                old.append(df)
+                n -= len(df)
+            else:
+                old.append(dfs[0].iloc[:n])
+                dfs[0] = dfs[0].iloc[n:]
+                n = 0
 
     return dfs, old
 
@@ -202,16 +204,21 @@ def diff_loc(dfs, new, window=None):
         List of dataframes to decay
     """
     dfs = deque(dfs)
-    dfs.append(new)
-    mx = max(df.index.max() for df in dfs)
-    mn = mx - pd.Timedelta(window)
+    if len(new) > 0:
+        dfs.append(new)
     old = []
-    while pd.Timestamp(dfs[0].index.min()) < mn:
-        o = dfs[0].loc[:mn]
-        old.append(o)  # TODO: avoid copy if fully lost
-        dfs[0] = dfs[0].iloc[len(o):]
-        if not len(dfs[0]):
-            dfs.popleft()
+    if len(dfs) > 0:
+        mx = max(df.index.max() for df in dfs)
+        mn = mx - pd.Timedelta(window) + pd.Timedelta('1ns')
+        while pd.Timestamp(dfs[0].index.min()) < mn:
+            o = dfs[0].loc[:mn]
+            if len(old) > 0:
+                old.append(o)
+            else:
+                old = [o]
+            dfs[0] = dfs[0].iloc[len(o):]
+            if not len(dfs[0]):
+                dfs.popleft()
 
     return dfs, old
 
@@ -337,7 +344,8 @@ def windowed_groupby_accumulator(acc, new, diff=None, window=None, agg=None, gro
 
     if 'groupers' in acc:
         groupers = deque(acc['groupers'])
-        groupers.append(grouper)
+        if len(grouper) > 0:
+            groupers.append(grouper)
         old_groupers, groupers = diff_align(dfs, groupers)
     else:
         old_groupers = [grouper] * len(old)

streamz/dataframe/tests/test_dataframes.py

@@ -755,12 +755,11 @@ def f(x):
     assert len(L) == 5
 
     first = df.iloc[:diff]
-    lost = first[first.index.min() + value:]
-    first = first.iloc[len(lost):]
+    first = first[first.index.max() - value + pd.Timedelta('1ns'):]
 
     assert_eq(L[0], f(first))
 
-    last = df.loc[index.max() - value + pd.Timedelta('1s'):]
+    last = df.loc[index.max() - value + pd.Timedelta('1ns'):]
 
     assert_eq(L[-1], f(last))
 
@@ -815,6 +814,42 @@ def f(x):
     assert_eq(L[-1], f(last))
 
 
+def test_windowing_value_empty_intermediate_index(stream):
+    def preprocess(df):
+        mask = df["amount"] == 5
+        df = df.loc[mask]
+        return df
+
+    source = stream.map(preprocess)
+
+    example = pd.DataFrame({"amount":[]})
+    sdf = DataFrame(stream=source, example=example)
+
+    output = sdf.window("2h").amount.sum().stream.gather().sink_to_list()
+
+    stream.emit(pd.DataFrame({"amount": [1, 2, 3]}, index=[pd.Timestamp("2050-01-01 00:00:00"),
+                                                           pd.Timestamp("2050-01-01 01:00:00"),
+                                                           pd.Timestamp("2050-01-01 02:00:00")]))
+
+    stream.emit(pd.DataFrame({"amount": [5, 5, 5]}, index=[pd.Timestamp("2050-01-01 03:00:00"),
+                                                           pd.Timestamp("2050-01-01 04:00:00"),
+                                                           pd.Timestamp("2050-01-01 05:00:00")]))
+
+    stream.emit(pd.DataFrame({"amount": [4, 5, 6]}, index=[pd.Timestamp("2050-01-01 06:00:00"),
+                                                           pd.Timestamp("2050-01-01 07:00:00"),
+                                                           pd.Timestamp("2050-01-01 08:00:00")]))
+
+    stream.emit(pd.DataFrame({"amount": [1, 2, 3]}, index=[pd.Timestamp("2050-01-01 09:00:00"),
+                                                           pd.Timestamp("2050-01-01 10:00:00"),
+                                                           pd.Timestamp("2050-01-01 11:00:00")]))
+
+    stream.emit(pd.DataFrame({"amount": [5, 5, 5]}, index=[pd.Timestamp("2050-01-01 12:00:00"),
+                                                           pd.Timestamp("2050-01-01 13:00:00"),
+                                                           pd.Timestamp("2050-01-01 14:00:00")]))
+
+    assert_eq(output, [0, 10, 5, 5, 10])
+
+
 def test_window_full():
     df = pd.DataFrame({'x': np.arange(10, dtype=float), 'y': [1.0, 2.0] * 5})