ML-585: writing empty df to nosql (#221)

integration/test_flow_integration.py

@@ -544,3 +544,13 @@ def test_cache_flushing(setup_teardown_test):
     controller.terminate()
     controller.await_termination()
 
+
+def test_write_empty_df(setup_teardown_test):
+    table = Table(setup_teardown_test, V3ioDriver())
+    df = pd.DataFrame({})
+
+    controller = build_flow([
+        DataframeSource(df),
+        NoSqlTarget(table),
+    ]).run()
+    controller.await_termination()

storey/table.py

@@ -363,8 +363,10 @@ async def _terminate(self):
             for key in self._changed_keys.copy():
                 if key not in self._pending_by_key:
                     await self._persist(_PersistJob(key, None, None), from_terminate=True)
-            await self._q.put(_termination_obj)
-            await self._worker_awaitable
+            if self._q:
+                # in case there was no _persist for this table, q and worker_awaitable were never created
+                await self._q.put(_termination_obj)
+                await self._worker_awaitable
             for value in self._attrs_cache.values():
                 value.lock = None
             self._q = None