{"record":{"id":"64fce60edf941164","repo":"pathwaycom/pathway","slug":"duplicated-entries-for-key-row-key","errorCode":null,"errorMessage":"duplicated entries for key {row.key}","messagePattern":"duplicated entries for key (.+?)","errorType":"validation","errorClass":"KeyError","httpStatus":null,"severity":"error","filePath":"python/pathway/internals/api.py","lineNumber":205,"sourceCode":"            raise ValueError(f\"Column {DIFF_PSEUDOCOLUMN} can only contain 1 and -1.\")\n        shard = data[SHARD_PSEUDOCOLUMN][i] if SHARD_PSEUDOCOLUMN in data else None\n        input_row = DataRow(\n            key, values, time=time, diff=diff, shard=shard, dtypes=dtypes\n        )\n        input_data.append(input_row)\n\n    return scope.static_table(input_data, connector_properties)\n\n\ndef squash_updates(\n    updates: CapturedStream, *, terminate_on_error: bool = True\n) -> CapturedTable:\n    state: CapturedTable = {}\n    updates.sort(key=lambda row: (row.time, row.diff))\n\n    def handle_error(row: DataRow, msg: str):\n        if terminate_on_error:\n            raise KeyError(msg)\n        else:\n            warnings.warn(msg)\n            t: tuple[Value, ...] = (ERROR,) * len(row.values)\n            state[row.key] = t\n\n    for row in updates:\n        if row.diff == 1:\n            if row.key in state:\n                handle_error(row, f\"duplicated entries for key {row.key}\")\n                continue\n            state[row.key] = tuple(row.values)\n        elif row.diff == -1:\n            if state[row.key] != tuple(row.values):\n                handle_error(row, f\"deleting non-existing entry {row.values}\")\n                continue\n            del state[row.key]\n        else:\n            handle_error(row, f\"invalid diff value: {row.diff}\")","sourceCodeStart":187,"sourceCodeEnd":223,"githubUrl":"https://github.com/pathwaycom/pathway/blob/fa2f74a4649b7c5908690cf60137263d8d80de5f/python/pathway/internals/api.py#L187-L223","documentation":"squash_updates folds a captured update stream into a table state; the semantics are that a key can be inserted (diff=1) at most once without a preceding deletion (diff=-1). When a second insert for a key that is already present arrives, the stream contradicts itself. By default terminate_on_error=True and this raises KeyError; with terminate_on_error=False the error is only warned about and the key's values become ERROR sentinels.","triggerScenarios":"Calling squash_updates (used internally when replaying captured/static data with updates) on a stream where the same key appears twice with diff=1 and no intervening diff=-1; generating test dataframes with duplicated primary keys and __diff__=1 on both rows; replaying a CDC feed where an UPDATE was encoded as two inserts instead of insert+delete.","commonSituations":"Synthetic test data generated with np.random so keys collide; CDC feeds where updates should be delete+insert pairs but were emitted as inserts; replaying deduplicated Kafka topics by primary key where duplicates slipped through.","solutions":["Deduplicate inserts by key before replay: keep only the first insert per (key, time) or convert later inserts into delete/insert pairs.","If updates are encoded as new inserts, emit diff=-1 for the old row before diff=1 for the new one.","For exploratory analysis where you accept corrupted keys, pass terminate_on_error=False and handle ERROR values downstream."],"exampleFix":"# before (duplicate inserts for key 'a')\nrows = [\n    (\"a\", 1, 1),\n    (\"a\", 2, 1),  # duplicated entries for key 'a'\n]\n\n# after (delete then re-insert)\nrows = [\n    (\"a\", 1, 1),\n    (\"a\", 1, -1),\n    (\"a\", 2, 1),\n]","handlingStrategy":"validation","validationCode":"seen: set = set()\nfor row in updates:\n    key, diff = row.key, row.diff\n    if diff == 1:\n        if key in seen:\n            raise ValueError(f\"duplicate insert for key {key}; add a diff=-1 row first\")\n        seen.add(key)\n    elif diff == -1:\n        seen.discard(key)","typeGuard":null,"tryCatchPattern":"try:\n    state = pw.internals.api.squash_updates(updates)\nexcept KeyError as e:\n    # duplicated insert for a key: fix the generator, don't continue with ERROR sentinels\n    raise RuntimeError(f\"update stream is inconsistent: {e}\") from e","preventionTips":["Validate captured streams for duplicate inserts before replaying them.","Generate synthetic test data with unique keys per insert, or explicit delete/insert pairs for updates.","Only use terminate_on_error=False in exploration; downstream code must handle ERROR values."],"tags":["data-ingestion","update-stream","primary-key","testing"],"backgroundTag":null,"analyzedSha":"fa2f74a4649b7c5908690cf60137263d8d80de5f","analyzedAt":"2026-08-15T01:48:17.006Z","schemaVersion":2},"datasetVersion":"2026-08-15T22:17:37.221Z"}