{"record":{"id":"6fc55ac94dad62b7","repo":"pathwaycom/pathway","slug":"unable-to-process-negative-update-with-this-statef","errorCode":null,"errorMessage":"Unable to process negative update with this stateful reducer.","messagePattern":"Unable to process negative update with this stateful reducer\\.","errorType":"exception","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"python/pathway/internals/custom_reducers.py","lineNumber":365,"sourceCode":"                    positive_updates.extend(_positive_updates)\n                    _positive_updates = []\n                    state = None\n                acc = Counter(positive_updates)\n                acc.subtract(negative_updates)\n                assert all(x >= 0 for x in acc.values())\n                positive_updates = list(acc.elements())\n                negative_updates = []\n\n            if state is None:\n                if neutral_available:\n                    state = reducer_cls.neutral()\n                    _positive_updates = []\n                    _cnt = 0\n                elif len(positive_updates) == 0:\n                    if len(negative_updates) == 0:\n                        return None\n                    else:\n                        raise ValueError(\n                            \"Unable to process negative update with this stateful reducer.\"\n                        )\n                else:\n                    if sort_by_available:\n                        positive_updates.sort(\n                            key=lambda x: reducer_cls.sort_by(list(x))\n                        )\n                    state = reducer_cls.from_row(list(positive_updates[0]))\n                    if not retract_available:\n                        _positive_updates = positive_updates[0:1]\n                        _cnt = 0\n                    else:\n                        _positive_updates = []\n                        _cnt = 1\n                    positive_updates = positive_updates[1:]\n\n            updates = [(row_up, False) for row_up in positive_updates] + [\n                (row_up, True) for row_up in negative_updates","sourceCodeStart":347,"sourceCodeEnd":383,"githubUrl":"https://github.com/pathwaycom/pathway/blob/fa2f74a4649b7c5908690cf60137263d8d80de5f/python/pathway/internals/custom_reducers.py#L347-L383","documentation":"Pathway wraps stateful custom reducers (classes with update/compute_result) for use in reduce() over tables with changing data. When the accumulator has no neutral() and no retract() implementation, it cannot represent or process row deletions. This instance fires when the state is still None (nothing accumulated yet) and the update batch contains only negative (retraction) rows — there is no way to apply a deletion to an empty/uninitialized state.","triggerScenarios":"Using a custom reducer via pw.reducers.udf_reducer(CustomCls) where CustomCls defines neither neutral() nor retract(), applied to a non-append-only input source (e.g. Kafka with updates/deletes, mutable CSVs) whose first batch for some group is purely retractions.","commonSituations":"Custom reducers tested only on static/debug tables (append-only) then deployed against streaming sources that emit deletions; windowed reductions over tables where forgetting produces retraction batches.","solutions":["Implement retract(self, other) on the accumulator class so deletions can be undone.","Alternatively implement neutral() so an identity state exists to retract from.","If the input is genuinely append-only, ensure it is declared/ingested as append-only so Pathway does not deliver negative updates.","Prefer a built-in reducer (sum, count, min...) which supports retractions when your aggregation permits it."],"exampleFix":"# before\nclass MySum(StatefulReducer):\n    def __init__(self): self.total = 0\n    def update(self, other): self.total += other\n    def compute_result(self): return self.total\n\n# after\nclass MySum(StatefulReducer):\n    def __init__(self): self.total = 0\n    def update(self, other): self.total += other\n    def retract(self, other): self.total -= other\n    def compute_result(self): return self.total","handlingStrategy":"validation","validationCode":"assert hasattr(ReducerCls, 'retract') or hasattr(ReducerCls, 'neutral'), 'custom reducer must support retractions (retract/neutral) on mutable sources'","typeGuard":"def reducer_handles_retractions(cls) -> bool:\n    return any('retract' in vars(c) or 'neutral' in vars(c) for c in cls.__mro__ if c is not object)","tryCatchPattern":null,"preventionTips":["Implement retract() in every custom reducer used on streaming inputs.","Test custom reducers with pw.debug tables containing retractions, not only append-only data."],"tags":["pathway","custom-reducers","streaming","retractions"],"backgroundTag":null,"analyzedSha":"fa2f74a4649b7c5908690cf60137263d8d80de5f","analyzedAt":"2026-08-15T01:48:17.006Z","schemaVersion":2},"datasetVersion":"2026-08-15T17:31:12.345Z"}