From 1ede8e39d82f0f4b517f6fde0ba32f2a63c2ea64 Mon Sep 17 00:00:00 2001 From: Gal Topper Date: Thu, 2 Nov 2023 17:05:48 +0800 Subject: [PATCH] ML-4879: Avoid keeping reference to events in `AggregateByKey` when cache is disabled (#463) Fixes [ML-4879](https://jira.iguazeng.com/browse/ML-4879). Co-authored-by: Gal Topper --- storey/table.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/storey/table.py b/storey/table.py index 7c1c8b9a..ee1dac98 100644 --- a/storey/table.py +++ b/storey/table.py @@ -190,7 +190,8 @@ async def _aggregate(self, key, event, data, timestamp): cache_item = self._get_aggregations_attrs(key) await cache_item.aggregate(data, timestamp) self._changed_keys.add(key) - self._pending_events.append(event) + if self._flush_task: + self._pending_events.append(event) async def _get_features(self, key, timestamp): if self._flush_exception is not None: