Add empty windowing index test and fix TODO

Author: chinmaychandak

streamz/dataframe/aggregations.py

@@ -212,7 +212,10 @@ def diff_loc(dfs, new, window=None):
         mn = mx - pd.Timedelta(window) + pd.Timedelta('1ns')
         while pd.Timestamp(dfs[0].index.min()) < mn:
             o = dfs[0].loc[:mn]
-            old.append(o)  # TODO: avoid copy if fully lost
+            if len(old) > 0:
+                old.append(o)
+            else:
+                old = [o]
             dfs[0] = dfs[0].iloc[len(o):]
             if not len(dfs[0]):
                 dfs.popleft()

streamz/dataframe/tests/test_dataframes.py

@@ -814,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})